Crypto Forem: Henry Lin

Hermes Agent 项目功能与通用使用场景分析

Henry Lin — Wed, 08 Apr 2026 13:06:09 +0000

Hermes Agent 项目功能与通用使用场景分析

1. 项目定位

Hermes Agent 是一个通用型 AI Agent 平台，不是单纯的聊天界面，也不是单一的代码助手。它把大模型推理、工具调用、终端执行、文件操作、网页检索、浏览器自动化、长期记忆、定时任务、多平台消息接入和外部系统扩展整合到一个统一框架里。

从仓库结构和主干代码来看，它的目标不是“回答问题”本身，而是让模型具备持续执行任务的能力，并且能在不同入口、不同环境和不同工具集之间稳定运行。

可以把它理解为三层系统：

Agent 执行层：负责会话循环、模型调用、工具调度、上下文压缩。
平台能力层：负责 CLI、消息网关、状态存储、调度、配置、权限边界。
扩展生态层：负责 skills、MCP、插件、自定义工具和多环境运行。

所以，Hermes Agent 更接近一个“可运行的 Agent Operating Layer”，而不是一个普通的 LLM 应用。

2. 项目主干结构

从代码组织看，几个核心模块职责很清楚：

run_agent.py
核心 AIAgent 实现，负责主对话循环、工具调用、消息历史和最终响应。
model_tools.py
工具发现与分发层。仓库里的工具模块在导入时注册到统一 registry，Agent 再按 toolset 组装可用能力。
tools/registry.py
工具注册中心，负责 schema、handler、可用性检测和 dispatch。
toolsets.py
定义不同场景下的工具组合，比如 CLI、gateway、API server、ACP 等。
cli.py 和 hermes_cli/
提供交互式终端 UI、slash command、模型切换、配置管理、setup、doctor、profile 等。
gateway/
提供 Telegram、Discord、Slack、WhatsApp、Signal、Email、Webhook 等消息平台入口。
hermes_state.py
使用 SQLite + WAL + FTS5 做会话持久化和全文检索。
cron/
提供定时任务能力，支持定时运行 agent 并把结果投递到配置的平台。
tools/mcp_tool.py
提供 MCP 接入，把外部系统能力注册成 Hermes 原生工具。
acp_adapter/
提供编辑器/ACP 集成能力。
batch_runner.py、environments/
提供批处理、轨迹生成和研究/训练相关能力。

从规模上看，这已经是一个完整平台：

tools/ 下有大量独立工具模块。
gateway/platforms/ 下有多种消息平台适配器。
tests/ 覆盖面很大，说明项目不是演示性质，而是实际维护中的系统。

3. 核心功能拆解

3.1 Agent 会话循环

Hermes 的核心是 AIAgent 的 tool-calling loop。基本模式是：

模型接收消息和可用工具定义。
模型决定是否调用工具。
Hermes 执行工具，并把结果写回消息历史。
循环直到模型产出最终答案，或者达到迭代限制。

这个设计的意义在于，Hermes 不是只让模型“建议怎么做”，而是让模型“真的去做”：

读取文件
搜索代码
执行命令
查询网页
调用浏览器
写入结果
再根据结果继续行动

这使它适合处理多步任务，而不是只做一轮问答。

3.2 工具注册与工具集

Hermes 的工具系统是项目的核心之一。

其特点是：

工具模块独立存在，一个文件通常负责一个工具域。
每个工具在导入时自注册到 registry。
Agent 不直接依赖具体工具，而是通过 toolset 组合决定当前会话能用什么。

这带来几个好处：

容易增删工具。
可以按平台控制权限边界。
不同入口可以有不同能力集合。
外部工具可以以统一方式接入。

从已有代码看，工具能力至少覆盖以下几类：

Web 搜索与网页抽取
终端执行与进程管理
文件读写、补丁和搜索
浏览器自动化
图像分析与生成
会话搜索
记忆与 TODO
代码执行沙箱
子代理委派
定时任务
跨平台消息发送
Home Assistant

这意味着 Hermes 不是“只有 shell 的 agent”，而是一个多能力编排器。

3.3 终端与多执行环境

tools/terminal_tool.py 表明 Hermes 支持多种命令执行后端，而不局限于本机 shell。

常见后端包括：

local
docker
ssh
modal
daytona
singularity

这使 Hermes 能适配不同部署模式：

本地开发机
远程服务器
容器化执行环境
云端弹性运行环境

对通用项目来说，这一点非常重要，因为很多 Agent 系统的问题不是“能不能推理”，而是“能不能在正确环境里做事”。Hermes 在这一层是有明确工程设计的。

3.4 CLI 交互系统

Hermes 的 CLI 不是一个简单的 input() 包装，而是一个完整的终端交互系统。

它具备：

多行输入
slash command
自动补全
tool progress 展示
会话切换和恢复
模型切换
配置修改
交互式 setup
doctor 诊断
皮肤和显示层定制

这意味着 Hermes 可以作为“长期使用的本地 Agent 工作台”，而不是一次性的命令入口。

3.5 消息网关

gateway/ 模块是 Hermes 与许多本地 agent 项目区别很大的地方。

它支持把同一个 agent 接到多种消息平台上，包括：

Telegram
Discord
Slack
WhatsApp
Signal
Email
Webhook
Home Assistant 等

这样带来的通用价值是：

Agent 不必绑定在终端里。
用户可以从不同平台继续同一个工作流。
自动任务结果可以直接投递到聊天平台。
Agent 可以成为团队协作接口，而不是个人玩具。

如果把 Hermes 放到服务器上，它更像一个“随处可用的个人/团队执行代理”。

3.6 会话持久化与检索

hermes_state.py 使用 SQLite + WAL + FTS5 管理 session 和 message。

这不是轻量日志，而是明确的长期状态存储设计，提供：

会话持久化
历史消息恢复
全文检索
标题化与检索式恢复
跨会话搜索

这类能力对长期使用的 Agent 非常关键，因为很多真正有价值的任务不是“当场回答”，而是：

记住上次做到哪里
找回过去解决过的问题
在多个 session 中复用历史上下文

Hermes 在这方面明显不是无状态聊天框架。

3.7 记忆与技能系统

Hermes 还有两个比较有辨识度的方向：

memory
skills

memory 的作用偏向长期事实沉淀，比如偏好、约束、常用模式。

skills 的作用偏向流程化知识沉淀，即：

某类任务怎么做
哪些步骤固定可复用
某个工作流如何标准化

从仓库设计看，skills 不是装饰功能，而是平台的一部分：

CLI 和 gateway 里有对应命令
prompt builder 会把相关指令纳入系统 prompt 逻辑
skill 文件体系是长期工作流复用的重要机制

这使 Hermes 更接近“能积累经验的 agent”，而不是每轮都从零开始。

3.8 上下文压缩与长任务支持

agent/context_compressor.py 显示 Hermes 对长对话场景有明确支持：

接近上下文上限时自动压缩
旧工具输出裁剪
结构化摘要
保留头尾关键内容

这很重要，因为真实任务里，很多 Agent 失败不是因为模型不会，而是因为：

对话太长
工具输出太多
历史信息失控

Hermes 在工程上显式处理了这个问题，因此更适合长流程任务。

3.9 子代理与程序化工具调用

Hermes 不只支持“一个 agent 干到底”，还支持：

delegate_task 子代理委派
execute_code 程序化工具调用

子代理的意义是把复杂问题拆开，让多个上下文隔离的 agent 并行处理。

程序化工具调用的意义是让模型先写一个 Python 脚本，再由脚本批量调用工具，通过 RPC 完成多步操作，减少模型回合数。

这种设计很适合：

需要并行调查多个问题
需要对多个对象执行重复步骤
需要压缩 token 成本
需要将步骤型操作转成脚本化流程

这说明 Hermes 在“任务执行效率”上做过比较深的思考。

3.10 定时任务与后台自动化

cron/ 模块使 Hermes 具备调度能力，而不是只能交互式运行。

典型能力包括：

定时运行 prompt
触发数据收集脚本
保存输出
推送结果到聊天平台
和 gateway 联动执行

这让 Hermes 具备“自动运行 agent job”的能力，适合做周期性报告、巡检、整理和提醒类任务。

3.11 MCP 与外部系统扩展

tools/mcp_tool.py 是 Hermes 变成“平台型系统”的关键。

它允许：

通过 stdio 或 HTTP 连接 MCP server
自动发现工具
把外部能力注入 Hermes 工具体系

这意味着 Hermes 不需要自己内置所有能力。它更像一个统一壳层，外部系统可以通过 MCP 接入，例如：

公司内部 API
数据平台
Issue 系统
文档系统
工单系统
DevOps 平台

这让 Hermes 的上限不再是仓库内置功能，而是可连接生态。

3.12 批处理与研究型能力

batch_runner.py 和相关环境目录说明 Hermes 还支持更偏研究/训练的能力：

批量运行 agent
保存 trajectory
统计工具调用情况
支持研究数据集驱动的运行方式

这部分对一般用户未必是核心入口，但对做 agent 评测、数据生成、研究实验的人来说很有价值。

4. 项目的通用能力画像

如果把上面这些功能浓缩成一句话，Hermes 的通用能力画像是：

“一个可长期运行、可跨平台接入、可调用多种工具、可持久化记忆与会话、可调度自动任务、可连接外部系统的通用 Agent 平台。”

更具体一点，它擅长以下五类事情：

把自然语言任务转成多步执行流程。
把多种工具和外部系统统一成一个可调用层。
把 agent 从终端扩展到聊天平台和后台任务。
把历史会话、技能和记忆沉淀成长期资产。
把一次性问答升级成持续运行的工作流。

5. 典型使用场景

以下场景不依赖任何特定行业，基本都适用于通用团队或个人用户。

5.1 个人长期助手

这是最直接的场景。

Hermes 可以作为个人长期助手，承担：

日常查询
文件整理
命令执行
提醒和定时任务
历史会话回顾
在 Telegram/Slack 上继续同一会话

和普通聊天助手相比，它的优势在于：

能执行动作
能记忆历史
能跨端继续
能定时自动运行

5.2 开发与调试助手

这是 Hermes 很强的适用面之一。

它适合：

阅读和搜索代码
修改文件
执行测试
分析日志
运行 shell 命令
编写脚本
使用子代理并行调查问题

尤其在大型代码库中，它比“只会生成代码的助手”更强，因为它能在本地环境真实执行和验证。

5.3 研究与资料整理助手

Hermes 很适合承担资料型任务：

搜索网页
抽取网页内容
汇总多来源信息
生成摘要
保存结果
周期性监控某类主题变化

如果配合 cron 和消息推送，可以形成自动研究简报工作流。

5.4 运维与巡检助手

通过 terminal、process、cron 和 messaging，Hermes 很适合做轻量运维助手：

定时检查服务状态
分析日志
发现异常后推送消息
执行标准化恢复动作
汇总巡检结果

它不一定适合作为高风险生产控制核心，但很适合作为巡检、观察、汇报和半自动处置层。

5.5 团队协作 Agent

由于 Hermes 有 gateway 和跨平台消息能力，它可以作为团队共享入口：

成员通过 Slack/Telegram 直接发起任务
Agent 拉取上下文并执行
结果回传到团队频道
定时任务自动发日报、周报、告警和摘要

这使它适合做团队内部的“智能机器人基础设施”。

5.6 文档与知识管理助手

Hermes 也适合做知识型工作：

搜索历史会话
归纳结论
生成知识文档
维护标准操作流程
把重复问题沉淀成 skills

相比只接数据库检索的知识机器人，它更强的地方在于它还能行动，能在整理知识时顺便验证、修正和生成产物。

5.7 自动化报告系统

这是 Hermes 很自然的落地形式。

例如：

每天整理新闻摘要
每周汇总项目状态
每小时巡检服务健康度
每晚生成工作进展报告

其关键能力来自：

cron
web / terminal / file 工具
持久化会话
messaging 投递

5.8 多系统统一入口

如果一个组织已经有多个内部系统，Hermes 很适合作为统一自然语言入口。

通过 MCP 或自定义工具，可以把这些系统统一接入：

文档平台
项目管理平台
CI/CD
工单系统
内部 API
知识库

这样用户就不必记住每个系统的入口和操作语法，而是通过 Hermes 统一调度。

6. 项目的工程价值

Hermes 的工程价值主要体现在以下方面。

6.1 它不是玩具式 agent

很多 agent 项目能 demo，但很难长期用。Hermes 明显考虑了长期运行问题：

状态持久化
工具分层
多入口
调度
配置体系
测试体系
权限和环境边界

6.2 它强调统一入口

CLI、gateway、ACP、API server、cron、MCP 这些模块放在一起，说明 Hermes 的设计目标不是单一产品，而是统一 agent runtime。

6.3 它强调可扩展性

registry、toolset、MCP、skills、plugins 这些机制说明 Hermes 的重点之一是“扩展”，不是“把一切写死在主循环里”。

6.4 它强调长期记忆和复用

session search、memory、skills、context compression 都在解决一个问题：

如何让 agent 不只在当前轮聪明，而是在长期使用中变得更有连续性。

7. 适合解决的问题类型

Hermes 特别适合下面这些任务：

多步骤、需要工具执行的任务
需要长期上下文的任务
需要跨平台访问的任务
需要周期性自动运行的任务
需要连接多个系统的任务
需要把经验沉淀成标准流程的任务

简单说，它适合“工作流型问题”，不只是“知识问答型问题”。

8. 不适合解决的问题类型

虽然 Hermes 很强，但也有明确边界。

8.1 不适合极低延迟场景

只要核心路径依赖大模型推理和多轮工具调用，就不适合：

毫秒级交互
超高频实时控制
严格时延敏感链路

8.2 不适合强确定性核心逻辑

Hermes 适合作为智能编排层，但不适合直接替代：

严格规则引擎
事务核心
高风险审批最终裁决器

8.3 不适合无边界的高权限自动执行

它可以执行命令、写文件、接外部系统，因此如果没有工具边界、审批机制和环境隔离，风险会很高。

项目本身其实已经考虑到这点，所以才有：

toolset
approval
environment backend
profile
配置隔离

9. 最适合的落地方式

如果从通用使用角度评价，Hermes 最适合以下三种落地方式。

9.1 本地个人工作台

适合开发者、研究者、重度终端用户。

特点：

直接用 CLI
连本地代码和 shell
有历史会话和技能沉淀

9.2 服务器上的长期运行 Agent

适合希望让 agent 常驻运行的用户。

特点：

部署到 VPS 或云主机
用 gateway 接 Telegram/Slack 等
用 cron 定时执行任务
用消息平台接收结果

9.3 组织内部的统一智能接口

适合中大型团队。

特点：

通过 MCP 接入内部系统
通过 gateway 提供统一入口
通过 skill 和 memory 沉淀组织经验

10. 最终评价

Hermes Agent 的核心价值不在于“模型本身多强”，而在于它把一个通用 agent 真正做成了可运行、可扩展、可持续使用的平台。

如果只看通用功能，它最突出的点有四个：

工具系统完整，能做真实任务执行。
多入口统一，既能本地用，也能在消息平台和后台运行。
长期状态能力强，支持会话、记忆、技能和上下文压缩。
扩展性好，能通过 MCP 和自定义工具连接外部世界。

因此，Hermes 最适合被理解为：

“面向长期任务执行和多系统编排的通用 Agent 平台。”

它不是最轻的方案，但如果目标是做一个真正能长期用、能接系统、能跑自动化、能在不同入口持续工作的 agent，Hermes 的设计是比较完整的。

第1章什么是 QuantConnect/Lean

Henry Lin — Sun, 22 Mar 2026 08:17:50 +0000

第1章什么是 Lean？

本章介绍 Lean 量化交易引擎的基本概念、核心功能以及 Lean 与 Lean CLI 的区别。

1.1 项目简介

Lean 是 QuantConnect 开源的事件驱动算法交易平台，专为量化交易开发者设计。它提供了完整的策略研究、历史回测、参数优化和实盘交易能力。

核心功能

功能模块	说明
策略研究	支持在 Jupyter Notebook 中进行数据探索和策略验证
历史回测	基于历史数据模拟交易策略表现
参数优化	自动搜索最优策略参数
实盘交易	支持多家券商的实盘接入
多资产类别	股票、期权、期货、外汇、加密货币等

支持的编程语言

C# - 原生语言，性能最佳
Python - 通过 Python.NET 集成，生态丰富

1.2 Lean vs Lean CLI

Lean Engine（本项目）

Lean Engine 是核心交易引擎，包含：

Lean/
├── Engine/              # 核心引擎
├── Algorithm/           # 算法基类
├── Algorithm.CSharp/    # C# 示例策略
├── Algorithm.Python/    # Python 示例策略
├── Algorithm.Framework/ # 框架化模型
├── Common/              # 通用组件
├── Brokerages/          # 券商接口
└── Launcher/            # 程序入口

Lean CLI（推荐用于策略开发）

Lean CLI 是命令行工具，简化了 Lean 的使用：

# 安装 CLI
pip install lean

# 创建项目
lean project-create

# 运行回测
lean backtest

# 启动研究环境
lean research

何时使用哪个？

场景	推荐工具
开发交易策略	Lean CLI
贡献 Lean 源码	Lean Engine
学习内部机制	Lean Engine
快速验证想法	Lean CLI

1.3 核心特性

事件驱动架构

Lean 采用事件驱动架构，算法响应各种事件：

from AlgorithmImports import *

class MyAlgorithm(QCAlgorithm):

    def Initialize(self):
        # 初始化设置
        self.AddEquity("SPY", Resolution.Minute)

    def OnData(self, data):
        # 每个数据点触发
        pass

    def OnOrderEvent(self, orderEvent):
        # 订单事件触发
        pass

    def OnEndOfDay(self, symbol):
        # 每日收盘时触发
        pass

多资产类别支持

资产类型	Add 方法	市场示例
股票	`AddEquity()`	USA, China, India
外汇	`AddForex()`	EURUSD, USDJPY
期货	`AddFuture()`	ES, NQ, CL
期权	`AddOption()`	SPY, AAPL
加密货币	`AddCrypto()`	BTCUSD, ETHUSD
CFD	`AddCfd()`	各种差价合约
指数	`AddIndex()`	SPX, VIX

模块化设计

Lean 的每个组件都可以自定义：

┌─────────────────────────────────────────────┐
│              QCAlgorithm                    │
│  ┌──────────────────────────────────────┐  │
│  │  Alpha Model        (交易信号生成)    │  │
│  │  Portfolio Model    (组合构建)        │  │
│  │  Execution Model    (订单执行)        │  │
│  │  Risk Model         (风险控制)        │  │
│  │  Universe Model     (证券选择)        │  │
│  └──────────────────────────────────────┘  │
│                                             │
│  ┌──────────────────────────────────────┐  │
│  │  Brokerage          (券商接口)        │  │
│  │  Data Feed          (数据源)          │  │
│  │  Fill Model         (成交模型)        │  │
│  │  Fee Model          (手续费模型)      │  │
│  │  Slippage Model     (滑点模型)        │  │
│  └──────────────────────────────────────┘  │
└─────────────────────────────────────────────┘

1.4 架构概览

                        ┌─────────────┐
                        │   用户输入   │
                        └──────┬──────┘
                               │
                        ┌──────▼──────┐
                        │  Launcher   │
                        └──────┬──────┘
                               │
        ┌──────────────────────┼──────────────────────┐
        │                      │                      │
┌───────▼────────┐    ┌────────▼────────┐    ┌───────▼────────┐
│  Algorithm     │    │   Data Feed     │    │   Transaction  │
│  (策略逻辑)    │    │  (数据提供器)    │    │   (交易处理)    │
└───────┬────────┘    └────────┬────────┘    └───────┬────────┘
        │                      │                      │
        └──────────────────────┼──────────────────────┘
                               │
                        ┌──────▼──────┐
                        │   Results    │
                        │  (结果输出)   │
                        └─────────────┘

1.5 本地与云端混合开发

Lean 支持本地开发和云端部署的混合模式：

┌─────────────┐         ┌─────────────┐
│  本地开发    │  <--->  │  云端平台    │
│  (Lean CLI) │         │(QuantConnect)│
└─────────────┘         └─────────────┘
     • 完整控制              • 数据获取
     • 调试方便              • 云计算资源
     • 数据隐私              • 策略分享

1.6 示例：第一个 Lean 算法

下面是一个最简单的 Lean 算法示例：

from AlgorithmImports import *

class HelloWorldAlgorithm(QCAlgorithm):
    """
    最简单的 Lean 算法示例
    功能：买入并持有 SPY
    """

    def Initialize(self):
        # 1. 设置回测时间范围
        self.SetStartDate(2023, 1, 1)    # 开始日期
        self.SetEndDate(2023, 12, 31)    # 结束日期
        self.SetCash(100000)             # 初始资金

        # 2. 添加证券
        self.AddEquity("SPY", Resolution.Daily)

        # 3. 设置benchmark
        self.SetBenchmark("SPY")

    def OnData(self, data):
        # 4. 如果还没有持仓，买入SPY
        if not self.Portfolio.Invested:
            self.SetHoldings("SPY", 1)   # 使用100%资金买入
            self.Debug("Purchased SPY")

# RD-Agent Tutorial - Chapter 3: Advanced Usage Techniques

Henry Lin — Sun, 22 Mar 2026 02:29:59 +0000

RD-Agent Tutorial - Chapter 3: Advanced Usage Techniques

3.1 Quantitative Finance Agent Deep Dive

Quantitative Finance Scenario Architecture

RD-Agent's quantitative finance agent RD-Agent(Q) is the industry's first data-driven multi-agent quantitative strategy development framework. It achieves full-stack quantitative research and development automation through coordinated factor-model joint optimization.

Core Architecture Design

graph TB
    A[Market Data] --> B[Data Preprocessing]
    B --> C[Factor Mining Agent]
    B --> D[Model Evolution Agent]
    C --> E[Factor Evaluation]
    D --> F[Model Evaluation]
    E --> G[Joint Optimization Agent]
    F --> G
    G --> H[Strategy Backtesting]
    H --> I[Performance Analysis]
    I --> J{Requirements Met?}
    J -->|No| K[Knowledge Feedback]
    K --> C
    K --> D
    J -->|Yes| L[Strategy Deployment]

    style C fill:#e1f5fe
    style D fill:#f3e5f5
    style G fill:#e8f5e8
    style H fill:#fff3e0

Qlib Framework Integration

RD-Agent(Q) deeply integrates with Microsoft's Qlib quantitative investment library, providing:

Data Management:

🗃️ Standardized financial data format
📈 Real-time and historical data interfaces
🔄 Automatic data update mechanism
🧹 Data cleaning and preprocessing

Backtesting Engine:

📊 High-performance vectorized backtesting
💰 Trading cost and slippage simulation
📋 Multiple portfolio construction strategies
📈 Risk metric calculation

Model Library:

🤖 Machine learning model integration
📊 Traditional quantitative factors
🧠 Deep learning models
🔄 Model ensemble strategies

Factor Mining and Optimization

`rdagent fin_factor` Explained

The factor mining agent focuses on discovering and optimizing effective quantitative factors.

Startup Command:

# Basic factor mining
rdagent fin_factor

# Specify configuration file
rdagent fin_factor --config factor_config.yaml

# Specify iteration count
export FACTOR_MAX_LOOP=15
rdagent fin_factor

Workflow:

Hypothesis Generation Phase

   # Generate factor hypotheses based on market theory and historical experience
   factor_hypotheses = [
       "technical_indicator_momentum_factor",
       "financial_quality_factor",
       "market_sentiment_factor",
       "macroeconomic_factor"
   ]

Factor Implementation Phase

   # Automatically generate factor calculation code
   class MomentumFactor:
       def calculate(self, data):
           # Auto-generated momentum factor calculation logic
           return (data['close'] / data['close'].shift(20) - 1)

Factor Evaluation Phase

   # Multi-dimensional factor evaluation
   evaluation_metrics = {
       'IC': 0.045,          # Information Coefficient
       'IC_IR': 1.2,         # Information Ratio
       'rank_IC': 0.038,     # Rank IC
       'turnover': 0.8,      # Turnover rate
       'max_drawdown': 0.15  # Maximum drawdown
   }

Factor Generation Strategies

Technical Analysis Factors:

technical_factors = {
    "momentum": [
        "price_momentum_5d", "price_momentum_20d",
        "volume_momentum", "volatility_momentum"
    ],
    "mean_reversion": [
        "rsi_divergence", "bollinger_position",
        "price_deviation"
    ],
    "trend": [
        "ma_trend", "macd_signal", "trend_strength"
    ]
}

Fundamental Factors:

fundamental_factors = {
    "profitability": [
        "roe_trend", "roa_improvement",
        "gross_margin_stability"
    ],
    "growth": [
        "revenue_growth_consistency", "eps_growth_quality",
        "cash_flow_growth"
    ],
    "valuation": [
        "pe_relative", "pb_sector_adjusted",
        "ev_ebitda_normalized"
    ]
}

Alternative Factors:

alternative_factors = {
    "sentiment": [
        "news_sentiment_score", "social_media_buzz",
        "analyst_revision_momentum"
    ],
    "network": [
        "supply_chain_strength", "industry_correlation",
        "peer_performance_influence"
    ]
}

Factor Effectiveness Evaluation

Evaluation Framework:

class FactorEvaluator:
    def __init__(self, benchmark_data, start_date, end_date):
        self.data = benchmark_data
        self.start_date = start_date
        self.end_date = end_date

    def evaluate_factor(self, factor_values):
        """Comprehensive factor effectiveness evaluation"""
        metrics = {}

        # 1. Information Coefficient analysis
        metrics['IC'] = self.calculate_IC(factor_values)
        metrics['IC_std'] = self.calculate_IC_stability(factor_values)
        metrics['IC_IR'] = metrics['IC'] / metrics['IC_std']

        # 2. Monotonicity test
        metrics['monotonicity'] = self.test_monotonicity(factor_values)

        # 3. Turnover rate analysis
        metrics['turnover'] = self.calculate_turnover(factor_values)

        # 4. Backtesting performance
        backtest_result = self.backtest_factor(factor_values)
        metrics.update(backtest_result)

        return metrics

Model Evolution and Optimization

`rdagent fin_model` Explained

The model evolution agent focuses on automated development and optimization of prediction models.

Supported Model Types:

Model Category	Specific Models	Applicable Scenarios	Features
Linear Models	LinearRegression, Ridge, Lasso	Simple fast prediction	High interpretability
Tree Models	XGBoost, LightGBM, CatBoost	Tabular data modeling	Feature importance
Neural Networks	MLP, TabNet, DeepFM	Complex relationship modeling	Non-linear fitting
Time Series Models	LSTM, GRU, Transformer	Sequence data prediction	Time dependency
Ensemble Models	Stacking, Voting, Blending	Improve prediction accuracy	Model fusion

Model Architecture Search

Automated Architecture Search:

class ModelArchitectureSearch:
    def __init__(self):
        self.search_space = {
            "n_layers": [2, 3, 4, 5],
            "hidden_dims": [64, 128, 256, 512],
            "dropout_rate": [0.1, 0.2, 0.3, 0.4],
            "activation": ["relu", "tanh", "gelu"],
            "optimizer": ["adam", "adamw", "sgd"],
            "learning_rate": [0.001, 0.01, 0.1]
        }

    def search_best_architecture(self, train_data, valid_data):
        """Search for optimal model architecture"""
        best_config = None
        best_score = float('-inf')

        for config in self.generate_configurations():
            model = self.build_model(config)
            score = self.evaluate_model(model, train_data, valid_data)

            if score > best_score:
                best_score = score
                best_config = config

        return best_config, best_score

Hyperparameter Optimization

Multiple Optimization Algorithms:

from optuna import create_study
from sklearn.model_selection import cross_val_score

class HyperparameterOptimizer:
    def __init__(self, model_class, param_space):
        self.model_class = model_class
        self.param_space = param_space

    def optimize_with_optuna(self, X_train, y_train, n_trials=100):
        """Hyperparameter optimization using Optuna"""
        def objective(trial):
            params = {}
            for param_name, param_range in self.param_space.items():
                if isinstance(param_range, list):
                    params[param_name] = trial.suggest_categorical(
                        param_name, param_range
                    )
                elif isinstance(param_range, tuple):
                    if isinstance(param_range[0], int):
                        params[param_name] = trial.suggest_int(
                            param_name, param_range[0], param_range[1]
                        )
                    else:
                        params[param_name] = trial.suggest_float(
                            param_name, param_range[0], param_range[1]
                        )

            model = self.model_class(**params)
            scores = cross_val_score(model, X_train, y_train, cv=5)
            return scores.mean()

        study = create_study(direction='maximize')
        study.optimize(objective, n_trials=n_trials)

        return study.best_params, study.best_value

Factor-Model Joint Optimization

`rdagent fin_quant` Explained

The joint optimization agent is the core innovation of RD-Agent(Q), implementing collaborative evolution of factors and models.

Joint Optimization Strategies:

Alternating Optimization Strategy

   class AlternatingOptimization:
       def __init__(self, factor_agent, model_agent):
           self.factor_agent = factor_agent
           self.model_agent = model_agent

       def optimize(self, max_iterations=10):
           for i in range(max_iterations):
               # Fix model, optimize factors
               new_factors = self.factor_agent.evolve_factors(
                   current_model=self.current_model
               )

               # Fix factors, optimize model
               new_model = self.model_agent.evolve_model(
                   current_factors=new_factors
               )

               # Evaluate joint performance
               performance = self.evaluate_joint_performance(
                   new_factors, new_model
               )

               if performance > self.best_performance:
                   self.update_best_solution(new_factors, new_model)

Multi-Objective Optimization Strategy

   class MultiObjectiveOptimization:
       def __init__(self):
           self.objectives = [
               "maximize_sharpe_ratio",
               "minimize_max_drawdown",
               "maximize_information_ratio",
               "minimize_turnover"
           ]

       def pareto_optimization(self, population):
           """Pareto front optimization"""
           pareto_front = []

           for individual in population:
               is_dominated = False
               for other in population:
                   if self.dominates(other, individual):
                       is_dominated = True
                       break

               if not is_dominated:
                   pareto_front.append(individual)

           return pareto_front

Multi-Objective Optimization Algorithms

NSGA-II Implementation:

class NSGA2Optimizer:
    def __init__(self, population_size=100, generations=50):
        self.population_size = population_size
        self.generations = generations

    def optimize(self, factor_space, model_space):
        # Initialize population
        population = self.initialize_population()

        for generation in range(self.generations):
            # Evaluate fitness
            fitness_values = self.evaluate_population(population)

            # Non-dominated sorting
            fronts = self.fast_non_dominated_sort(population, fitness_values)

            # Select next generation
            next_population = self.select_next_generation(
                fronts, population, fitness_values
            )

            # Crossover and mutation
            offspring = self.crossover_and_mutation(next_population)
            population = offspring

        return self.get_pareto_front(population)

Collaborative Evolution Mechanism

Knowledge Sharing Mechanism:

class KnowledgeSharing:
    def __init__(self):
        self.factor_knowledge_base = {}
        self.model_knowledge_base = {}
        self.interaction_patterns = {}

    def share_factor_insights(self, factor_id, insights):
        """Share factor insights"""
        self.factor_knowledge_base[factor_id] = {
            'effectiveness': insights['ic_score'],
            'stability': insights['ic_std'],
            'best_models': insights['compatible_models'],
            'market_regimes': insights['effective_periods']
        }

    def share_model_insights(self, model_id, insights):
        """Share model insights"""
        self.model_knowledge_base[model_id] = {
            'architecture': insights['model_config'],
            'performance': insights['validation_score'],
            'best_factors': insights['important_features'],
            'hyperparameters': insights['optimal_params']
        }

    def get_recommendations(self, current_factors, current_model):
        """Get optimization recommendations"""
        recommendations = {
            'factor_suggestions': [],
            'model_suggestions': [],
            'joint_strategies': []
        }

        # Generate recommendations based on historical knowledge
        for factor in current_factors:
            if factor in self.factor_knowledge_base:
                knowledge = self.factor_knowledge_base[factor]
                recommendations['model_suggestions'].extend(
                    knowledge['best_models']
                )

        return recommendations

Financial Report Analysis

`rdagent fin_factor_report` Usage

Automatically extracting quantitative factors from financial reports is a unique capability of RD-Agent.

Supported Report Types:

📋 Annual Reports (10-K, Annual Report)
📊 Quarterly Reports (10-Q, Quarterly Report)
📰 News Announcements (8-K, Press Release)
📈 Analyst Reports (Research Report)
🏛️ Regulatory Filings (SEC Filings)

Usage Example:

# Basic report analysis
rdagent fin_factor_report --report-folder ./financial_reports

# Specify report type
rdagent fin_factor_report \
    --report-folder ./reports \
    --report-type annual \
    --language zh

# Batch processing
rdagent fin_factor_report \
    --report-folder ./reports \
    --batch-size 50 \
    --parallel-workers 4

Report Parsing and Feature Extraction

Text Preprocessing Pipeline:

class ReportPreprocessor:
    def __init__(self):
        self.nlp = spacy.load("en_core_web_sm")
        self.financial_terms = self.load_financial_dictionary()

    def preprocess_report(self, report_text):
        """Preprocess financial report"""
        # 1. Document cleaning
        cleaned_text = self.clean_document(report_text)

        # 2. Section segmentation
        sections = self.extract_sections(cleaned_text)

        # 3. Entity recognition
        entities = self.extract_financial_entities(sections)

        # 4. Sentiment analysis
        sentiment_scores = self.analyze_sentiment(sections)

        return {
            'sections': sections,
            'entities': entities,
            'sentiment': sentiment_scores,
            'metadata': self.extract_metadata(report_text)
        }

Factor Extraction Strategy:

class FinancialFactorExtractor:
    def __init__(self):
        self.factor_extractors = {
            'growth_factors': GrowthFactorExtractor(),
            'quality_factors': QualityFactorExtractor(),
            'sentiment_factors': SentimentFactorExtractor(),
            'risk_factors': RiskFactorExtractor()
        }

    def extract_factors(self, processed_report):
        """Extract factors from processed report"""
        extracted_factors = {}

        for factor_type, extractor in self.factor_extractors.items():
            factors = extractor.extract(processed_report)
            extracted_factors[factor_type] = factors

        return extracted_factors

class GrowthFactorExtractor:
    def extract(self, report_data):
        """Extract growth factors"""
        growth_factors = {}

        # Revenue growth quality
        revenue_mentions = self.find_revenue_discussions(report_data)
        growth_factors['revenue_growth_quality'] = self.analyze_growth_quality(
            revenue_mentions
        )

        # Market expansion strategy
        expansion_mentions = self.find_expansion_discussions(report_data)
        growth_factors['expansion_strategy_score'] = self.score_expansion_strategy(
            expansion_mentions
        )

        # R&D investment commitment
        rd_mentions = self.find_rd_discussions(report_data)
        growth_factors['rd_commitment_score'] = self.analyze_rd_commitment(
            rd_mentions
        )

        return growth_factors

Sentiment Analysis and Signal Mining

Multi-dimensional Sentiment Analysis:

class FinancialSentimentAnalyzer:
    def __init__(self):
        self.sentiment_model = pipeline(
            "sentiment-analysis",
            model="ProsusAI/finbert"
        )
        self.uncertainty_detector = UncertaintyDetector()
        self.forward_looking_detector = ForwardLookingDetector()

    def analyze_comprehensive_sentiment(self, text_sections):
        """Comprehensive sentiment analysis"""
        sentiment_scores = {}

        for section_name, text in text_sections.items():
            # Basic sentiment analysis
            basic_sentiment = self.sentiment_model(text)

            # Uncertainty detection
            uncertainty_score = self.uncertainty_detector.detect(text)

            # Forward-looking statement detection
            forward_looking = self.forward_looking_detector.detect(text)

            sentiment_scores[section_name] = {
                'polarity': basic_sentiment[0]['score'],
                'label': basic_sentiment[0]['label'],
                'uncertainty': uncertainty_score,
                'forward_looking': forward_looking,
                'confidence': self.calculate_confidence(
                    basic_sentiment, uncertainty_score
                )
            }

        return sentiment_scores

class SignalGenerator:
    def __init__(self):
        self.signal_weights = {
            'management_tone': 0.3,
            'financial_health': 0.4,
            'market_position': 0.2,
            'future_outlook': 0.1
        }

    def generate_trading_signals(self, factor_scores, sentiment_scores):
        """Generate trading signals"""
        composite_score = 0.0

        # Weighted composite score
        for factor_type, weight in self.signal_weights.items():
            if factor_type in factor_scores:
                composite_score += factor_scores[factor_type] * weight

        # Sentiment adjustment
        sentiment_adjustment = self.calculate_sentiment_adjustment(
            sentiment_scores
        )

        final_score = composite_score * sentiment_adjustment

        # Generate signal
        if final_score > 0.6:
            signal = "STRONG_BUY"
        elif final_score > 0.3:
            signal = "BUY"
        elif final_score > -0.3:
            signal = "HOLD"
        elif final_score > -0.6:
            signal = "SELL"
        else:
            signal = "STRONG_SELL"

        return {
            'signal': signal,
            'confidence': abs(final_score),
            'score': final_score,
            'components': {
                'factor_score': composite_score,
                'sentiment_adjustment': sentiment_adjustment
            }
        }

3.2 CoSTEER Evolution Framework Deep Dive

Evolution Framework Core Principles

CoSTEER (Collaborative evolving STrategy for automatic data-cEntric dEvelopment fRamework) is RD-Agent's core evolution engine, implementing autonomous learning and continuous improvement for agents.

Collaborative Evolution Strategy Design

The CoSTEER framework is based on the following core principles:

Multi-Agent Collaboration - Different specialized agents work together
Knowledge Accumulation - Learn and improve from historical experience
Adaptive Evolution - Dynamically adjust strategies based on feedback
RAG Enhancement - Retrieval-augmented generation improves decision quality

class CoSTEER(Developer[Experiment]):
    """CoSTEER evolution framework core class"""

    def __init__(
        self,
        settings: CoSTEERSettings,
        eva: RAGEvaluator,
        es: EvolvingStrategy,
        evolving_version: int = 2,
        with_knowledge: bool = True,
        knowledge_self_gen: bool = True,
        max_loop: int | None = None,
    ):
        self.settings = settings
        self.max_loop = settings.max_loop if max_loop is None else max_loop
        self.knowledge_base_path = Path(settings.knowledge_base_path)
        self.with_knowledge = with_knowledge
        self.knowledge_self_gen = knowledge_self_gen
        self.evolving_strategy = es
        self.evaluator = eva

        # Initialize RAG system
        self.rag_strategy = self._init_rag_strategy(evolving_version)

Knowledge Management Mechanism

Multi-level Knowledge Representation:

class KnowledgeManagementSystem:
    def __init__(self):
        self.knowledge_layers = {
            'factual_knowledge': FactualKnowledgeBase(),
            'procedural_knowledge': ProceduralKnowledgeBase(),
            'experiential_knowledge': ExperientialKnowledgeBase(),
            'meta_knowledge': MetaKnowledgeBase()
        }

    def store_knowledge(self, knowledge_item):
        """Layered knowledge storage"""
        knowledge_type = self.classify_knowledge(knowledge_item)
        self.knowledge_layers[knowledge_type].store(knowledge_item)

    def retrieve_relevant_knowledge(self, query_context):
        """Retrieve relevant knowledge"""
        relevant_knowledge = {}

        for layer_name, knowledge_base in self.knowledge_layers.items():
            relevant_items = knowledge_base.query(
                query_context,
                similarity_threshold=0.7
            )
            relevant_knowledge[layer_name] = relevant_items

        return self.synthesize_knowledge(relevant_knowledge)

Knowledge Representation Format:

@dataclass
class KnowledgeItem:
    """Knowledge item data structure"""
    id: str
    content: str
    knowledge_type: KnowledgeType
    domain: str
    confidence: float
    source: str
    timestamp: datetime
    usage_count: int = 0
    success_rate: float = 0.0

    def to_embedding(self) -> np.ndarray:
        """Convert to vector representation"""
        return self.embedding_model.encode(self.content)

    def update_effectiveness(self, success: bool):
        """Update knowledge effectiveness"""
        self.usage_count += 1
        if success:
            self.success_rate = (
                (self.success_rate * (self.usage_count - 1) + 1.0)
                / self.usage_count
            )
        else:
            self.success_rate = (
                self.success_rate * (self.usage_count - 1)
                / self.usage_count
            )

Adaptive Optimization Algorithm

Multi-Strategy Evolution Algorithm:

class AdaptiveEvolutionStrategy:
    def __init__(self):
        self.evolution_strategies = {
            'genetic_algorithm': GeneticAlgorithmStrategy(),
            'differential_evolution': DifferentialEvolutionStrategy(),
            'particle_swarm': ParticleSwarmStrategy(),
            'simulated_annealing': SimulatedAnnealingStrategy()
        }
        self.strategy_performance = {}
        self.adaptive_weights = {}

    def evolve_population(self, population, generation):
        """Adaptive population evolution"""
        # Select evolution strategy
        selected_strategies = self.select_strategies(generation)

        new_population = []
        for strategy_name, weight in selected_strategies.items():
            strategy = self.evolution_strategies[strategy_name]

            # Allocate population by weight
            sub_population_size = int(len(population) * weight)
            sub_population = population[:sub_population_size]

            # Evolve sub-population
            evolved_sub_pop = strategy.evolve(sub_population)
            new_population.extend(evolved_sub_pop)

        # Update strategy performance
        self.update_strategy_performance(new_population, generation)

        return new_population

    def select_strategies(self, generation):
        """Select evolution strategy based on historical performance"""
        if generation < 5:
            # Uniform distribution in early stages
            return {name: 1.0/len(self.evolution_strategies)
                   for name in self.evolution_strategies.keys()}

        # Adjust weights based on performance
        total_performance = sum(self.strategy_performance.values())
        return {
            name: performance / total_performance
            for name, performance in self.strategy_performance.items()
        }

Custom Evolution Strategies

Strategy Configuration and Adjustment

Configuration File Example:

# costeer_config.yaml
costeer_settings:
  max_loop: 10
  knowledge_base_path: "./knowledge_base"
  new_knowledge_base_path: "./new_knowledge"
  evolving_version: 2

evolution_strategy:
  population_size: 50
  mutation_rate: 0.1
  crossover_rate: 0.8
  selection_method: "tournament"
  tournament_size: 5

rag_settings:
  embedding_model: "text-embedding-3-small"
  similarity_threshold: 0.75
  max_retrieved_docs: 10
  rerank_top_k: 5

evaluation_settings:
  metrics: ["accuracy", "f1_score", "auc_roc"]
  cross_validation_folds: 5
  test_ratio: 0.2

knowledge_management:
  auto_cleanup: true
  cleanup_threshold: 0.3
  knowledge_update_frequency: 100
  max_knowledge_items: 10000

Custom Evolution Strategy Implementation:

class CustomEvolutionStrategy(EvolvingStrategy):
    """Custom evolution strategy"""

    def __init__(self, config: dict):
        super().__init__()
        self.config = config
        self.population_size = config.get('population_size', 50)
        self.mutation_rate = config.get('mutation_rate', 0.1)
        self.crossover_rate = config.get('crossover_rate', 0.8)

    def initialize_population(self, size: int = None):
        """Initialize population"""
        size = size or self.population_size
        population = []

        for _ in range(size):
            individual = self.create_random_individual()
            population.append(individual)

        return population

    def evolve_generation(self, population, fitness_scores):
        """Evolve one generation"""
        new_population = []

        # Elite selection - preserve best individuals
        elite_count = int(len(population) * 0.1)
        elite_indices = np.argsort(fitness_scores)[-elite_count:]
        for idx in elite_indices:
            new_population.append(population[idx])

        # Generate new individuals through crossover and mutation
        while len(new_population) < len(population):
            # Select parents
            parent1 = self.tournament_selection(population, fitness_scores)
            parent2 = self.tournament_selection(population, fitness_scores)

            # Crossover
            if random.random() < self.crossover_rate:
                child1, child2 = self.crossover(parent1, parent2)
            else:
                child1, child2 = parent1, parent2

            # Mutation
            if random.random() < self.mutation_rate:
                child1 = self.mutate(child1)
            if random.random() < self.mutation_rate:
                child2 = self.mutate(child2)

            new_population.extend([child1, child2])

        return new_population[:len(population)]

    def crossover(self, parent1, parent2):
        """Crossover operation"""
        # Implement specific crossover logic
        # Parameter crossover example here
        child1_params = {}
        child2_params = {}

        for key in parent1.parameters.keys():
            if random.random() < 0.5:
                child1_params[key] = parent1.parameters[key]
                child2_params[key] = parent2.parameters[key]
            else:
                child1_params[key] = parent2.parameters[key]
                child2_params[key] = parent1.parameters[key]

        child1 = parent1.copy()
        child2 = parent2.copy()
        child1.parameters = child1_params
        child2.parameters = child2_params

        return child1, child2

    def mutate(self, individual):
        """Mutation operation"""
        mutated = individual.copy()

        for key, value in mutated.parameters.items():
            if random.random() < self.mutation_rate:
                if isinstance(value, (int, float)):
                    # Gaussian mutation
                    noise = np.random.normal(0, 0.1 * abs(value))
                    mutated.parameters[key] = value + noise
                elif isinstance(value, str):
                    # String mutation (e.g., model names)
                    mutated.parameters[key] = self.mutate_string(value)

        return mutated

Evaluator Customization

Multi-Metric Evaluator:

class MultiMetricEvaluator(RAGEvaluator):
    """Multi-metric evaluator"""

    def __init__(self, metrics_config: dict):
        super().__init__()
        self.metrics_config = metrics_config
        self.metric_weights = metrics_config.get('weights', {})
        self.metric_calculators = self._init_metric_calculators()

    def evaluate(self, experiment: Experiment) -> Dict[str, float]:
        """Multi-metric evaluation"""
        metrics = {}

        for metric_name, calculator in self.metric_calculators.items():
            try:
                score = calculator.calculate(experiment)
                metrics[metric_name] = score
            except Exception as e:
                logger.warning(f"Metric {metric_name} calculation failed: {e}")
                metrics[metric_name] = 0.0

        # Calculate weighted composite score
        weighted_score = self._calculate_weighted_score(metrics)
        metrics['weighted_score'] = weighted_score

        return metrics

    def _calculate_weighted_score(self, metrics: Dict[str, float]) -> float:
        """Calculate weighted composite score"""
        total_score = 0.0
        total_weight = 0.0

        for metric_name, score in metrics.items():
            if metric_name in self.metric_weights:
                weight = self.metric_weights[metric_name]
                total_score += score * weight
                total_weight += weight

        return total_score / total_weight if total_weight > 0 else 0.0

class CustomMetricCalculator:
    """Custom metric calculator"""

    def __init__(self, metric_name: str, calculation_func: callable):
        self.metric_name = metric_name
        self.calculation_func = calculation_func

    def calculate(self, experiment: Experiment) -> float:
        """Calculate metric value"""
        return self.calculation_func(experiment)

# Usage example
def sharpe_ratio_calculator(experiment):
    """Calculate Sharpe ratio"""
    returns = experiment.get_returns()
    if len(returns) == 0 or returns.std() == 0:
        return 0.0
    return returns.mean() / returns.std() * np.sqrt(252)

def max_drawdown_calculator(experiment):
    """Calculate maximum drawdown"""
    cumulative_returns = (1 + experiment.get_returns()).cumprod()
    running_max = cumulative_returns.expanding().max()
    drawdown = (cumulative_returns - running_max) / running_max
    return -drawdown.min()  # Return positive value

# Create custom evaluator
metrics_config = {
    'weights': {
        'sharpe_ratio': 0.4,
        'max_drawdown': 0.3,
        'information_ratio': 0.3
    }
}

evaluator = MultiMetricEvaluator(metrics_config)
evaluator.add_metric_calculator('sharpe_ratio', sharpe_ratio_calculator)
evaluator.add_metric_calculator('max_drawdown', max_drawdown_calculator)

Knowledge Base Management

Knowledge Base Optimization Strategy:

class KnowledgeBaseManager:
    """Knowledge base manager"""

    def __init__(self, config: dict):
        self.config = config
        self.knowledge_store = {}
        self.embedding_cache = {}
        self.access_frequency = defaultdict(int)
        self.last_access_time = {}

    def add_knowledge(self, knowledge_item: KnowledgeItem):
        """Add knowledge item"""
        # Check knowledge base size limit
        if len(self.knowledge_store) >= self.config.get('max_items', 10000):
            self._cleanup_knowledge()

        self.knowledge_store[knowledge_item.id] = knowledge_item
        self.embedding_cache[knowledge_item.id] = knowledge_item.to_embedding()
        self.last_access_time[knowledge_item.id] = datetime.now()

    def query_knowledge(
        self,
        query: str,
        top_k: int = 10,
        similarity_threshold: float = 0.7
    ) -> List[KnowledgeItem]:
        """Query relevant knowledge"""
        query_embedding = self._get_query_embedding(query)
        similarities = {}

        for item_id, item_embedding in self.embedding_cache.items():
            similarity = cosine_similarity(
                query_embedding.reshape(1, -1),
                item_embedding.reshape(1, -1)
            )[0][0]

            if similarity >= similarity_threshold:
                similarities[item_id] = similarity
                self.access_frequency[item_id] += 1
                self.last_access_time[item_id] = datetime.now()

        # Sort by similarity and return top-k
        sorted_items = sorted(
            similarities.items(),
            key=lambda x: x[1],
            reverse=True
        )[:top_k]

        return [self.knowledge_store[item_id] for item_id, _ in sorted_items]

    def _cleanup_knowledge(self):
        """Clean up knowledge base"""
        cleanup_count = int(len(self.knowledge_store) * 0.1)  # Clean 10%

        # Composite score based on access frequency and time
        scores = {}
        current_time = datetime.now()

        for item_id, knowledge_item in self.knowledge_store.items():
            frequency_score = self.access_frequency[item_id]
            time_score = (
                current_time - self.last_access_time[item_id]
            ).days
            effectiveness_score = knowledge_item.success_rate

            # Composite score (high frequency, recent, effective = high score)
            composite_score = (
                frequency_score * 0.4 +
                effectiveness_score * 0.4 -
                time_score * 0.2
            )
            scores[item_id] = composite_score

        # Remove lowest scored items
        items_to_remove = sorted(
            scores.items(),
            key=lambda x: x[1]
        )[:cleanup_count]

        for item_id, _ in items_to_remove:
            del self.knowledge_store[item_id]
            del self.embedding_cache[item_id]
            del self.access_frequency[item_id]
            del self.last_access_time[item_id]

        logger.info(f"Cleaned up {cleanup_count} knowledge items")

Multi-Agent Collaboration Mechanism

Research Agent Responsibilities

class ResearchAgent:
    """Research agent - responsible for hypothesis generation and strategy planning"""

    def __init__(self, knowledge_base: KnowledgeBase):
        self.knowledge_base = knowledge_base
        self.hypothesis_generator = HypothesisGenerator()
        self.strategy_planner = StrategyPlanner()

    def generate_hypotheses(self, problem_context: dict) -> List[Hypothesis]:
        """Generate research hypotheses"""
        # Retrieve relevant theories from knowledge base
        relevant_knowledge = self.knowledge_base.query(
            problem_context['description']
        )

        # Generate hypotheses based on domain knowledge
        hypotheses = []
        for knowledge_item in relevant_knowledge:
            hypothesis = self.hypothesis_generator.generate_from_knowledge(
                knowledge_item, problem_context
            )
            hypotheses.append(hypothesis)

        # Generate novel hypotheses
        novel_hypotheses = self.hypothesis_generator.generate_novel(
            problem_context, existing_hypotheses=hypotheses
        )
        hypotheses.extend(novel_hypotheses)

        return self.rank_hypotheses(hypotheses)

    def plan_research_strategy(self, hypothesis: Hypothesis) -> ResearchPlan:
        """Plan research strategy"""
        return self.strategy_planner.create_plan(
            hypothesis=hypothesis,
            available_resources=self.get_available_resources(),
            time_constraints=self.get_time_constraints()
        )

Development Agent Functions

class DevelopmentAgent:
    """Development agent - responsible for code implementation and testing validation"""

    def __init__(self, code_generator: CodeGenerator, validator: CodeValidator):
        self.code_generator = code_generator
        self.validator = validator
        self.implementation_cache = {}

    def implement_hypothesis(self, hypothesis: Hypothesis) -> Implementation:
        """Implement research hypothesis"""
        # Generate implementation code
        implementation_code = self.code_generator.generate_code(
            specification=hypothesis.specification,
            constraints=hypothesis.constraints
        )

        # Validate code correctness
        validation_result = self.validator.validate(implementation_code)

        if not validation_result.is_valid:
            # Fix code issues
            fixed_code = self.code_generator.fix_issues(
                implementation_code,
                validation_result.issues
            )
            implementation_code = fixed_code

        # Create implementation object
        implementation = Implementation(
            code=implementation_code,
            hypothesis=hypothesis,
            validation_result=validation_result
        )

        # Cache implementation
        self.implementation_cache[hypothesis.id] = implementation

        return implementation

    def optimize_implementation(
        self,
        implementation: Implementation,
        feedback: Feedback
    ) -> Implementation:
        """Optimize implementation based on feedback"""
        optimization_suggestions = self.analyze_feedback(feedback)

        optimized_code = self.code_generator.optimize_code(
            implementation.code,
            optimization_suggestions
        )

        return Implementation(
            code=optimized_code,
            hypothesis=implementation.hypothesis,
            parent_implementation=implementation
        )

Inter-Agent Communication Protocol

class AgentCommunicationProtocol:
    """Inter-agent communication protocol"""

    def __init__(self):
        self.message_queue = {}
        self.subscriptions = defaultdict(list)
        self.message_handlers = {}

    def register_agent(self, agent_id: str, agent: Agent):
        """Register agent"""
        self.message_queue[agent_id] = []
        agent.set_communication_protocol(self)

    def subscribe(self, agent_id: str, message_type: str):
        """Subscribe to message type"""
        self.subscriptions[message_type].append(agent_id)

    def publish(self, sender_id: str, message_type: str, message: dict):
        """Publish message"""
        # Add message metadata
        full_message = {
            'sender': sender_id,
            'type': message_type,
            'timestamp': datetime.now(),
            'content': message
        }

        # Send to all subscribers
        for subscriber_id in self.subscriptions[message_type]:
            if subscriber_id != sender_id:  # Don't send to self
                self.message_queue[subscriber_id].append(full_message)

    def get_messages(self, agent_id: str) -> List[dict]:
        """Get agent's messages"""
        messages = self.message_queue.get(agent_id, [])
        self.message_queue[agent_id] = []  # Clear queue
        return messages

class CollaborativeWorkflow:
    """Collaborative workflow"""

    def __init__(self, agents: List[Agent], protocol: AgentCommunicationProtocol):
        self.agents = {agent.id: agent for agent in agents}
        self.protocol = protocol
        self.workflow_state = WorkflowState.INITIALIZED

    def execute_collaborative_task(self, task: CollaborativeTask):
        """Execute collaborative task"""
        # Decompose task
        subtasks = self.decompose_task(task)

        # Assign subtasks to suitable agents
        task_assignments = self.assign_tasks(subtasks)

        # Execute subtasks
        results = {}
        for agent_id, assigned_tasks in task_assignments.items():
            agent = self.agents[agent_id]
            agent_results = agent.execute_tasks(assigned_tasks)
            results[agent_id] = agent_results

        # Coordinate and integrate results
        integrated_result = self.integrate_results(results)

        return integrated_result

    def coordinate_agents(self):
        """Coordinate agent activities"""
        # Process inter-agent messages
        for agent_id, agent in self.agents.items():
            messages = self.protocol.get_messages(agent_id)
            if messages:
                agent.process_messages(messages)

        # Check if coordination is needed
        coordination_needed = self.check_coordination_needs()
        if coordination_needed:
            self.perform_coordination()

This chapter demonstrates RD-Agent's advanced features and deep customization capabilities, including specialized functions of the quantitative finance agent, core mechanisms of the CoSTEER evolution framework, and implementation methods for multi-agent collaboration. The next chapter will demonstrate practical applications of these advanced features through real-world case studies.

# RD-Agent Tutorial - Chapter 2: Core Functions

Henry Lin — Sun, 22 Mar 2026 02:29:38 +0000

RD-Agent Tutorial - Chapter 2: Core Functions

2.1 Command Line Interface Basics

CLI Structure and Design Principles

RD-Agent adopts a unified command line interface (CLI) design, with all functions accessed through the rdagent command. The CLI core implementation is located at rdagent/app/cli.py, built using the Typer framework, providing an intuitive and powerful command line experience.

Architecture Analysis

# rdagent/app/cli.py core structure
import typer
from rdagent.app.data_science.loop import main as data_science
from rdagent.app.qlib_rd_loop.factor import main as fin_factor
from rdagent.app.qlib_rd_loop.model import main as fin_model
from rdagent.app.qlib_rd_loop.quant import main as fin_quant

app = typer.Typer()

# Register commands
app.command(name="fin_factor")(fin_factor)
app.command(name="fin_model")(fin_model)
app.command(name="data_science")(data_science)

Command Classification and Organization

RD-Agent commands are organized by functional domain:

Command Category	Command	Function Description	Primary Use
Financial Quant	`fin_factor`	Automatic factor mining	Discover effective quantitative factors
	`fin_model`	Automatic model evolution	Optimize prediction models
	`fin_quant`	Factor-model joint optimization	End-to-end strategy development
	`fin_factor_report`	Report factor extraction	Extract signals from financial reports
Data Science	`data_science`	General data science tasks	ML competitions, modeling projects
General Model	`general_model`	Paper model implementation	Research paper reproduction
Utility Functions	`ui`	Web interface	Visualization and monitoring
	`health_check`	Health check	System status verification
	`collect_info`	Information collection	Diagnostics and debugging

Automatic Environment Loading Mechanism

An important feature of the RD-Agent CLI is automatic environment variable loading:

from dotenv import load_dotenv
load_dotenv(".env")  # Automatically load .env file from current directory

Advantages:

🔄 Automatic configuration loading, no manual environment variable setup needed
📁 Supports project-level configuration, different projects can have different configs
🔒 Configuration files stored locally, improved security

Basic Command Usage

Help System Usage

# View all available commands
rdagent --help

# View detailed help for specific commands
rdagent fin_factor --help
rdagent data_science --help

# View subcommand parameters
rdagent ui --help

Example Output:

Usage: rdagent [OPTIONS] COMMAND [ARGS]...

Options:
  --help  Show this message and exit.

Commands:
  data_science        Run data science automation
  fin_factor         Run factor mining
  fin_model          Run model evolution
  fin_quant          Run factor-model joint optimization
  general_model      Extract and implement models from papers
  health_check       Check system health
  ui                 Start web interface

Global Parameters and Options

While most of RD-Agent's configuration is managed through .env files, certain commands support runtime parameters:

# Common parameters for UI command
rdagent ui --port 19899 --log-dir ./logs --debug

# Selective checks for health check
rdagent health_check --no-check-env --no-check-docker

# Competition specification for data science command
rdagent data_science --competition tabular-playground-series-dec-2021

Logging and Output Management

RD-Agent provides comprehensive logging capabilities:

# Run with default log level
rdagent fin_factor

# Enable verbose logging (via environment variable)
export RDAGENT_LOG_LEVEL=DEBUG
rdagent fin_factor

# Specify log output directory
export RDAGENT_LOG_DIR=./custom_logs
rdagent fin_quant

Log Structure:

logs/
├── rdagent.log                 # Main log file
├── experiments/                # Experiment logs
│   ├── factor_exp_001/
│   └── model_exp_001/
└── ui/                        # Web UI logs
    └── access.log

Monitoring Interface Usage

Web UI Startup and Configuration

RD-Agent provides a powerful web interface for real-time monitoring and result viewing:

# Basic startup
rdagent ui --port 19899

# Specify log directory
rdagent ui --port 19899 --log-dir ./logs

# Enable debug mode
rdagent ui --port 19899 --debug

# Data science specific interface
rdagent ui --port 19899 --data_science True

Real-time Monitoring Features

The Web UI provides the following monitoring capabilities:

1. Experiment Progress Tracking

📊 Real-time experiment status display
📈 Performance metric charts
⏱️ Timeline view
🔄 Auto-refresh

2. Log Viewer

📝 Structured log display
🔍 Log search and filtering
📋 Multi-level logs (INFO, DEBUG, ERROR)
💾 Log export functionality

3. Result Visualization

📊 Factor performance comparison
📈 Model training curves
🎯 Backtest result charts
📋 Detailed performance reports

Interface Function Details

Main Dashboard:

┌─────────────────────────────────────────┐
│ RD-Agent Monitoring Panel                │
├─────────────────────────────────────────┤
│ 🔄 Experiment Status: Running (2/5)      │
│ ⏱️ Runtime: 1h 23m                       │
│ 📊 Current Task: Factor Validation       │
│ 🎯 Best Score: 0.234 (IC)                │
└─────────────────────────────────────────┘

Experiment List:

Experiment ID | Type   | Status   | Start Time | Score
--------------|--------|----------|------------|-------
EXP_001       | Factor | Complete | 14:23:01   | 0.198
EXP_002       | Factor | Running  | 14:45:12   | -
EXP_003       | Model  | Waiting  | -          | -

2.2 Data Science Agent Details

Data Science Scenario Overview

RD-Agent's data science agent is a fully automated machine learning engineering system capable of autonomously completing the entire workflow from data exploration to model deployment. This agent achieved first place in the MLE-bench benchmark, demonstrating powerful automation capabilities.

Supported Task Types

Task Type	Description	Application Scenarios	Technical Features
Tabular Data Modeling	Prediction tasks for structured data	Financial prediction, user behavior analysis	Feature engineering, model ensembling
Time Series Forecasting	Predictive modeling for time series data	Stock price prediction, demand forecasting	Time series features, sequence models
Image Classification	Computer vision tasks	Medical imaging, product recognition	CNN, transfer learning
Natural Language Processing	Text data processing	Sentiment analysis, document classification	Pretrained models, embeddings
Regression Analysis	Continuous value prediction	Price prediction, rating estimation	Linear/non-linear models

Workflow and Architecture

graph TD
    A[Data Input] --> B[Hypothesis Generation]
    B --> C[Experiment Design]
    C --> D[Feature Engineering]
    D --> E[Model Development]
    E --> F[Validation Testing]
    F --> G[Result Analysis]
    G --> H{Performance Satisfied?}
    H -->|No| I[Feedback Learning]
    I --> B
    H -->|Yes| J[Model Output]

    style B fill:#e1f5fe
    style D fill:#f3e5f5
    style E fill:#e8f5e8
    style G fill:#fff3e0

Core Components:

Hypothesis Generator - Generate modeling hypotheses based on data features and problem type
Experiment Manager - Systematically manage experiment workflow and resources
Feature Engineer - Automatically generate and select effective features
Model Developer - Automatically select and tune models
Evaluation System - Multi-dimensional assessment of model performance
Knowledge Manager - Accumulate and reuse experiment experience

Integration with Other Tools

# Supported data formats and sources
Data Sources:
├── CSV/Excel files
├── Kaggle competition data
├── Database connections (SQL)
├── API data sources
├── Image folders
└── Text document collections

Model Frameworks:
├── Scikit-learn
├── XGBoost/LightGBM
├── PyTorch
├── TensorFlow/Keras
├── Transformers (Hugging Face)
└── Custom models

Kaggle Competition Automation

Kaggle API Configuration

First, configure the Kaggle API to access competition data:

# 1. Get Kaggle API Token
# Login to Kaggle -> Account -> Create New Token
# Download kaggle.json file

# 2. Configure API file
mkdir -p ~/.kaggle
mv ~/Downloads/kaggle.json ~/.kaggle/
chmod 600 ~/.kaggle/kaggle.json

# 3. Verify configuration
kaggle competitions list

Environment Variable Configuration

Add data science related configuration to your .env file:

# Data science agent configuration
DS_LOCAL_DATA_PATH="./data/competitions"
DS_CODER_ON_WHOLE_PIPELINE=True
DS_IF_USING_MLE_DATA=True
DS_SAMPLE_DATA_BY_LLM=True
DS_SCEN=rdagent.scenarios.data_science.scen.KaggleScen

# LLM configuration (as mentioned earlier)
CHAT_MODEL=gpt-4o
EMBEDDING_MODEL=text-embedding-3-small
OPENAI_API_KEY=your-api-key

Automatic Competition Data Download

# Automatically download and process competition data
rdagent data_science --competition tabular-playground-series-dec-2021

# Specify custom data path
export DS_LOCAL_DATA_PATH="./custom_data"
rdagent data_science --competition house-prices-advanced-regression-techniques

Complete Competition Workflow Demonstration

Using "Tabular Playground Series - Dec 2021" as an example:

Step 1: Environment Setup

# Create project directory
mkdir kaggle_rdagent_demo && cd kaggle_rdagent_demo

# Configure environment
cat > .env << EOF
CHAT_MODEL=gpt-4o
EMBEDDING_MODEL=text-embedding-3-small
OPENAI_API_KEY=your-openai-api-key

DS_LOCAL_DATA_PATH="$(pwd)/data"
DS_CODER_ON_WHOLE_PIPELINE=True
DS_IF_USING_MLE_DATA=True
DS_SAMPLE_DATA_BY_LLM=True
DS_SCEN=rdagent.scenarios.data_science.scen.KaggleScen
EOF

Step 2: Launch Automation Workflow

# Start RD-Agent data science agent
rdagent data_science --competition tabular-playground-series-dec-2021

Step 3: Real-time Monitoring

# Start monitoring interface in another terminal
rdagent ui --port 19899 --data_science True
# Visit http://localhost:19899

Automatically Executed Workflow:

Data Download and Exploration
- Automatically download competition data
- Generate data exploration report
- Identify data types and distributions
Hypothesis Generation
- Generate modeling hypotheses based on data features
- Analyze target variable characteristics
- Identify potential feature engineering opportunities
Feature Engineering
- Automatic numerical feature transformations
- Categorical feature encoding
- Interaction feature creation
- Time feature extraction (if applicable)
Model Development
- Try multiple algorithms (random forest, XGBoost, neural networks, etc.)
- Automatic hyperparameter tuning
- Cross-validation and model selection
Model Ensembling
- Stacking/Blending strategies
- Multi-model fusion
- Weight optimization
Submission Preparation
- Generate prediction files
- Format validation
- Automatic submission (optional)

Medical Prediction Model Scenario

Medical Data Processing

Medical data typically has special characteristics, and RD-Agent provides specialized processing capabilities:

# Run medical prediction task
rdagent data_science --competition medical-prediction-task

# Example: Acute kidney failure prediction
wget https://github.com/SunsetWolf/rdagent_resource/releases/download/ds_data/arf-12-hours-prediction-task.zip
unzip arf-12-hours-prediction-task.zip -d ./data/

Medical Data Characteristics Processing:

Privacy Protection
- Data anonymization and de-identification
- HIPAA compliance considerations
- Local processing, no raw data upload
Imbalanced Data Handling
- SMOTE oversampling
- Cost-sensitive learning
- Threshold tuning
Time Series Medical Data
- Disease course time modeling
- Multi-timepoint features
- Survival analysis methods
Multimodal Data Fusion
- Structured data (lab indicators)
- Unstructured data (medical record text)
- Medical image data

Prediction Model Development

# Example of special configuration for medical prediction
medical_config = {
    "task_type": "classification",
    "positive_class_weight": 10,  # Handle imbalance
    "cross_validation_folds": 10,  # More folds
    "feature_selection": "medical_relevance",
    "interpretability": True,  # Medical requires interpretability
    "ensemble_methods": ["voting", "stacking"],
    "evaluation_metrics": ["auc", "precision", "recall", "f1"],
}

Model Evaluation and Optimization

Medical model evaluation focuses more on:

🎯 Clinical Metrics: Sensitivity, Specificity, PPV, NPV
📊 ROC/PR Curves: Performance at different thresholds
⚖️ Cost-Benefit Analysis: Cost of misdiagnosis and missed diagnosis
🔍 Interpretability: SHAP values and feature importance

General Data Science Tasks

Custom Dataset Processing

For non-competition data, RD-Agent supports flexible data integration:

# Create data directory structure
mkdir -p custom_project/data
cd custom_project

# Prepare data files
# train.csv - Training data
# test.csv - Test data (optional)
# sample_submission.csv - Submission format (optional)

# Configure environment
cat > .env << EOF
DS_LOCAL_DATA_PATH="$(pwd)/data"
DS_CODER_ON_WHOLE_PIPELINE=True
DS_IF_USING_MLE_DATA=False  # Non-MLE data
DS_SAMPLE_DATA_BY_LLM=False
DS_SCEN=rdagent.scenarios.data_science.scen.DataScienceScen
EOF

# Run analysis
rdagent data_science --competition custom_project

Data Format Requirements

Standard Tabular Data:

data/
├── train.csv          # Required: Training data
├── test.csv           # Optional: Test data
├── sample_submission.csv  # Optional: Submission format
└── description.md     # Optional: Problem description

Multimodal Data:

data/
├── tabular/
│   ├── train.csv
│   └── test.csv
├── images/
│   ├── train/
│   └── test/
├── text/
│   ├── train_texts.json
│   └── test_texts.json
└── config.yaml       # Data configuration file

Model Selection and Tuning

RD-Agent intelligently selects the most suitable algorithms:

Classification Task Algorithm Selection:

classification_algorithms = {
    "small_data": ["RandomForest", "SVM", "LogisticRegression"],
    "medium_data": ["XGBoost", "LightGBM", "CatBoost"],
    "large_data": ["NeuralNetwork", "TabNet", "AutoML"],
    "text_data": ["BERT", "RoBERTa", "DistilBERT"],
    "image_data": ["ResNet", "EfficientNet", "ViT"],
}

Hyperparameter Tuning Strategies:

tuning_strategies = {
    "bayesian_optimization": "Efficient parameter space search",
    "random_search": "Fast preliminary tuning",
    "grid_search": "Precise but time-consuming search",
    "evolutionary": "Global optimization for complex spaces",
    "hyperband": "Multi-fidelity optimization",
}

Automated Feature Engineering

Numerical Feature Processing:

numerical_transformations = [
    "StandardScaler",      # Standardization
    "MinMaxScaler",        # Normalization
    "RobustScaler",        # Robust scaling
    "PowerTransformer",    # Power transform
    "QuantileTransformer", # Quantile transform
    "PCA",                 # Principal component analysis
    "PolynomialFeatures",  # Polynomial features
]

Categorical Feature Processing:

categorical_transformations = [
    "OneHotEncoder",       # One-hot encoding
    "LabelEncoder",        # Label encoding
    "TargetEncoder",       # Target encoding
    "BinaryEncoder",       # Binary encoding
    "HashingEncoder",      # Hash encoding
    "FrequencyEncoder",    # Frequency encoding
]

Automatic Feature Generation:

auto_feature_generation = [
    "feature_interactions",  # Feature interactions
    "feature_aggregations", # Feature aggregations
    "feature_ratios",       # Feature ratios
    "feature_differences",  # Feature differences
    "time_based_features",  # Time-based features
    "text_embeddings",      # Text embeddings
]

Summary

This chapter provided detailed coverage of RD-Agent's core functions, including:

Command Line Interface Basics - Unified CLI design and monitoring capabilities
Data Science Agent Details - Complete automated machine learning workflow
Practical Application Scenarios - Kaggle competitions, medical prediction, general modeling

The next chapter will delve into the advanced features of quantitative finance agents and general model agents.

Quick Reference Commands:

# Start data science project
rdagent data_science --competition <project_name>

# Start monitoring interface
rdagent ui --port 19899 --data_science True

# Health check
rdagent health_check

# View help
rdagent data_science --help

RD-Agent Tutorial - Chapter 1: Basic Introduction

Henry Lin — Sun, 22 Mar 2026 02:29:22 +0000

RD-Agent Tutorial - Chapter 1: Basic Introduction

1.1 Project Overview and Core Philosophy

Introduction to RD-Agent

RD-Agent (Research & Development Agent) is an open-source automated R&D framework developed by Microsoft Research, specifically designed for building data-driven AI solutions. This project achieved first place in the industry-standard machine learning engineering benchmark MLE-bench, making it the best-performing machine learning agent to date.

Core Philosophy: The R&D Framework

The design philosophy of RD-Agent is based on two core components:

R (Research): 🔬 Research Component
- Automatically propose new hypotheses and ideas
- Analyze limitations of existing solutions
- Generate innovative solutions based on domain knowledge
D (Development): 🛠️ Development Component
- Transform research ideas into executable code
- Implement and validate hypotheses
- Continuously optimize and improve solutions

MLE-bench Benchmark Performance

Agent	Low (%)	Medium (%)	High (%)	All (%)
R&D-Agent o1-preview	48.18 ± 2.49	8.95 ± 2.36	18.67 ± 2.98	22.4 ± 1.1
R&D-Agent o3(R)+GPT-4.1(D)	51.52 ± 6.21	7.89 ± 3.33	16.67 ± 3.65	22.45 ± 2.45
AIDE o1-preview	34.3 ± 2.4	8.8 ± 1.1	10.0 ± 1.9	16.9 ± 1.1

Core Advantages and Application Value

1. Automated Data-Driven R&D Process

End-to-end Automation: Complete workflow from data analysis to model deployment
Intelligent Decision Support: Automatically generate solutions based on data insights
Continuous Learning: Learn from historical experience and continuously improve

2. Multi-Agent Collaboration Architecture

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Research Agent │    │ Development     │    │  Evaluation      │
│                 │◄──►│ Agent           │◄──►│ Agent           │
│• Hypothesis      │    │• Code           │    │• Performance    │
│  generation      │    │  implementation   │    │  evaluation     │
│• Strategy        │    │• Testing        │    │• Feedback       │
│  planning        │    │• Optimization   │    │• quality control │
└─────────────────┘    └─────────────────┘    └─────────────────┘

3. Continuous Learning and Evolution Capabilities

CoSTEER Framework: Collaborative Strategy Evolution Framework
Knowledge Accumulation: Automatically build and maintain domain knowledge base
Adaptive Optimization: Dynamically adjust strategies based on feedback

Application Scenarios and Target Users

🏢 Quantitative Investment Teams

Automatic Factor Mining: Discover effective factors from massive data
Strategy Development: Automatically generate and optimize trading strategies
Risk Management: Intelligent risk control and portfolio optimization
Cost Efficiency: Achieve 2x returns at less than $10 cost compared to traditional methods

🔬 Data Scientists

Automated Feature Engineering: Intelligent feature generation and selection
Model Selection Optimization: Automated model architecture search
Experiment Management: Systematic experiment tracking and comparison
Pipeline Construction: End-to-end machine learning pipeline auto-construction

🎓 Researchers and Scholars

Paper Reproduction: Automatically extract and implement models from papers
Research Assistant: Intelligent literature analysis and knowledge extraction
Experiment Design: Scientific experiment planning and execution
Research Documentation: Automated research achievement documentation

🏆 Kaggle Competition Participants

End-to-end Automation: From data exploration to model submission
Feature Engineering: Intelligent feature creation and combination
Model Ensemble: Multi-model fusion strategy optimization
Time Management: Efficient competition time allocation

1.2 System Requirements and Environment Preparation

Hardware and Software Requirements

Operating System Support

Linux: ✅ Fully supported (Ubuntu 18.04+, CentOS 7+)
Windows: ❌ Currently not supported
macOS: ❌ Currently not supported

Important Note: RD-Agent currently supports only Linux systems. If you are using Windows or macOS, it is recommended to use Docker, WSL2, or a virtual machine to run a Linux environment.

Python Version Requirements

Python 3.10: ✅ Fully supported (recommended)
Python 3.11: ✅ Fully supported
Python 3.9: ⚠️ Some features may be limited
Python 3.12+: ⚠️ May have compatibility issues

Memory and Computing Resource Recommendations

Use Case	Memory Requirement	CPU Requirement	Storage Space
Lightweight Use	4GB+	2 cores+	10GB+
Medium Scale	8GB+	4 cores+	50GB+
Large Scale	16GB+	8 cores+	100GB+
Production	32GB+	16 cores+	500GB+

Docker Environment Configuration

Docker Installation and Configuration

RD-Agent heavily relies on Docker to provide isolated runtime environments, ensuring experiment reproducibility.

Ubuntu/Debian Docker Installation:

# Update package index
sudo apt-get update

# Install necessary packages
sudo apt-get install \
    apt-transport-https \
    ca-certificates \
    curl \
    gnupg \
    lsb-release

# Add Docker official GPG key
curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg --dearmor -o /usr/share/keyrings/docker-archive-keyring.gpg

# Set up stable repository
echo \
  "deb [arch=amd64 signed-by=/usr/share/keyrings/docker-archive-keyring.gpg] https://download.docker.com/linux/ubuntu \
  $(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null

# Install Docker Engine
sudo apt-get update
sudo apt-get install docker-ce docker-ce-cli containerd.io

CentOS/RHEL Docker Installation:

# Install necessary packages
sudo yum install -y yum-utils

# Add Docker repository
sudo yum-config-manager \
    --add-repo \
    https://download.docker.com/linux/centos/docker-ce.repo

# Install Docker Engine
sudo yum install docker-ce docker-ce-cli containerd.io

# Start Docker service
sudo systemctl start docker
sudo systemctl enable docker

Running Docker Without sudo Permissions

This is a required configuration step! RD-Agent requires that the current user can run Docker commands without using sudo.

# Create docker user group (if it doesn't exist)
sudo groupadd docker

# Add current user to docker group
sudo usermod -aG docker $USER

# Reload user group permissions (or re-login)
newgrp docker

# Verify configuration is successful
docker run hello-world

Expected Output:

Hello from Docker!
This message shows that your installation appears to be working correctly.
...

LLM API Preparation

Supported LLM Providers

RD-Agent supports multiple large language model providers through LiteLLM:

Provider	Supported Models	Recommended Use	Cost Considerations
OpenAI	GPT-4, GPT-4-turbo, GPT-3.5-turbo	General purpose, high quality output	$$$
Azure OpenAI	GPT-4, GPT-3.5-turbo	Enterprise applications, compliance requirements	$$$
DeepSeek	deepseek-chat, deepseek-coder	Code generation, cost-effective	$
Anthropic	Claude-3.x	Long text processing, analysis tasks	$$
Google	Gemini Pro	Multimodal requirements	$$

API Key Acquisition Guide

OpenAI API Key Acquisition:

Visit OpenAI API Official Site
Register/login to your account
Go to "API Keys" page
Click "Create new secret key"
Copy and securely save the API key

Azure OpenAI Configuration:

Login to Azure Portal
Create Azure OpenAI resource
Get API key, endpoint, and deployment name
Record API version information

DeepSeek API Key Acquisition:

Visit DeepSeek API Platform
Register account and complete verification
Go to API key management page
Generate new API key

Cost Budget Recommendations

Usage Intensity	Monthly Budget	Applicable Scenarios
Learning & Trial	$10-50	Tutorial learning, small-scale experiments
Medium Usage	$50-200	Project development, medium-scale applications
Heavy Usage	$200-1000	Production environment, large-scale applications
Enterprise	$1000+	Enterprise deployment, continuous operation

1.3 Installation and Basic Configuration

Installation Method Selection

User Version Installation (Recommended)

For most users, installing directly via PyPI is the simplest approach:

# Create conda environment (recommended)
conda create -n rdagent python=3.10
conda activate rdagent

# Or use venv
python3.10 -m venv rdagent_env
source rdagent_env/bin/activate

# Install RD-Agent
pip install rdagent

Developer Version Installation

If you want to experience the latest features or participate in development, install from source:

# Clone repository
git clone https://github.com/microsoft/RD-Agent.git
cd RD-Agent

# Create development environment
conda create -n rdagent-dev python=3.10
conda activate rdagent-dev

# Install development dependencies
make dev

# Or manual installation
pip install -e .
pip install -r requirements/dev.txt

Dependency Management and Virtual Environments

Why do you need a virtual environment?

Avoid package version conflicts
Ensure environment consistency
Facilitate environment management and migration

Recommended environment management methods:

# Using conda (recommended)
conda create -n rdagent python=3.10
conda activate rdagent

# Or use pyenv + venv
pyenv install 3.10.12
pyenv local 3.10.12
python -m venv rdagent_env
source rdagent_env/bin/activate

Basic Environment Configuration

`.env` File Configuration Details

RD-Agent uses .env files to manage environment variable configurations. Create a .env file in the project root directory:

# Create .env file in project directory
touch .env

LiteLLM Unified Backend Configuration

Method 1: Unified API Basic Configuration (Recommended for Beginners)

# OpenAI configuration example
CHAT_MODEL=gpt-4o
EMBEDDING_MODEL=text-embedding-3-small
OPENAI_API_KEY=sk-your-openai-api-key-here
OPENAI_API_BASE=https://api.openai.com/v1

Method 2: Separated API Configuration (Advanced Users)

# Chat model configuration
CHAT_MODEL=gpt-4o
OPENAI_API_KEY=sk-your-openai-api-key-here
OPENAI_API_BASE=https://api.openai.com/v1

# Embedding model configuration (using different provider)
EMBEDDING_MODEL=litellm_proxy/BAAI/bge-large-en-v1.5
LITELLM_PROXY_API_KEY=your-siliconflow-api-key
LITELLM_PROXY_API_BASE=https://api.siliconflow.cn/v1

Multiple LLM Provider Configuration Examples

OpenAI Standard Configuration:

# Basic configuration
CHAT_MODEL=gpt-4o
EMBEDDING_MODEL=text-embedding-3-small
OPENAI_API_KEY=sk-proj-xxxxxxxxxxxxxxxxxxxxxxxxxx
OPENAI_API_BASE=https://api.openai.com/v1

# Optional: Special reasoning model configuration
# REASONING_THINK_RM=True  # Enable thinking process removal

Azure OpenAI Configuration:

# Azure OpenAI configuration
CHAT_MODEL=azure/gpt-4o-deployment-name
EMBEDDING_MODEL=azure/embedding-deployment-name
AZURE_API_KEY=your-azure-api-key
AZURE_API_BASE=https://your-resource.openai.azure.com/
AZURE_API_VERSION=2024-02-15-preview

DeepSeek Cost-Effective Configuration:

# DeepSeek chat model (cost-effective)
CHAT_MODEL=deepseek/deepseek-chat
DEEPSEEK_API_KEY=sk-your-deepseek-api-key

# SiliconFlow embedding model (DeepSeek has no embedding model)
EMBEDDING_MODEL=litellm_proxy/BAAI/bge-m3
LITELLM_PROXY_API_KEY=sk-your-siliconflow-api-key
LITELLM_PROXY_API_BASE=https://api.siliconflow.cn/v1

# Reasoning model configuration
REASONING_THINK_RM=True

Mixed Configuration (Recommended for Advanced Users):

# High-performance chat model
CHAT_MODEL=gpt-4o
OPENAI_API_KEY=sk-your-openai-api-key

# Cost-effective embedding model
EMBEDDING_MODEL=litellm_proxy/BAAI/bge-large-en-v1.5
LITELLM_PROXY_API_KEY=sk-your-siliconflow-api-key
LITELLM_PROXY_API_BASE=https://api.siliconflow.cn/v1

# Other configuration
REASONING_THINK_RM=False

Health Check and Verification

`rdagent health_check` Usage

After installation, perform a system health check first:

# Complete health check
rdagent health_check

# Skip environment variable check
rdagent health_check --no-check-env

# Skip Docker check
rdagent health_check --no-check-docker

# Only check port usage
rdagent health_check --no-check-env --no-check-docker

Expected Successful Output:

✅ Docker is installed and working correctly
✅ Port 19899 is available
✅ Environment variables are configured correctly
✅ LLM API connection successful
✅ All systems check passed!

Common Configuration Issue Troubleshooting

Issue 1: Docker Permission Error

❌ Error: Got permission denied while trying to connect to the Docker daemon socket

Solution:

# Ensure user is in docker group
sudo usermod -aG docker $USER
newgrp docker
# Re-run health check
rdagent health_check

Issue 2: Invalid API Key

❌ Error: Invalid API key provided

Solution:

# Check API key in .env file
cat .env | grep API_KEY
# Ensure key is correct and valid
# Reconfigure API key

Issue 3: Port Already in Use

❌ Error: Port 19899 is already in use

Solution:

# Check port usage
netstat -tlnp | grep 19899
# Kill occupying process or change port
rdagent ui --port 19900

Issue 4: Network Connection Problem

❌ Error: Failed to connect to API endpoint

Solution:

# Check network connectivity
curl -I https://api.openai.com/v1/models
# Configure proxy (if needed)
export https_proxy=http://proxy.company.com:8080
export http_proxy=http://proxy.company.com:8080

Environment Verification Checklist

Before starting to use RD-Agent, please confirm the following checklist:

[ ] Operating System: Linux environment
[ ] Python Version: 3.10 or 3.11
[ ] Virtual Environment: Dedicated environment activated
[ ] Docker: Installed and can run without sudo
[ ] Network Connection: Can access LLM API endpoints
[ ] API Keys: Valid with sufficient balance
[ ] Storage Space: At least 10GB available space
[ ] Port: Port 19899 available (or other specified port)
[ ] Permissions: Current user has read/write permissions to project directory

You have now completed the basic installation and configuration of RD-Agent. The next chapter will detail the core functions and usage methods of RD-Agent.

Quick Verification of Installation

After completing all configurations, let's perform a quick functional verification:

# Activate environment
conda activate rdagent

# Verify installation
python -c "import rdagent; print('RD-Agent installed successfully!')"

# Run health check
rdagent health_check

# View available commands
rdagent --help

If all steps complete successfully, you are now ready to start using the powerful features of RD-Agent!

工业级量化开源软件 QuantConnect/LEAN 功能介绍

Henry Lin — Sat, 21 Mar 2026 12:20:59 +0000

如果你在寻找一个能够承载“机构级”逻辑的量化框架，QuantConnect/LEAN 是目前开源界最重工业级、功能最全的选择。

相比你之前关注的 NautilusTrader（侧重高性能 Rust/Python 混合）或 Alphalens（侧重因子分析），LEAN 是一个完整的闭环生态系统。它不仅是回测引擎，更是一套包含了订单管理（OMS）、数据处理、风险控制以及多语言支持的复杂工程。

以下是 LEAN 的核心功能深度分析：

1. 核心架构：多语言与事件驱动

C# 核心，Python 友好： LEAN 底层使用 C# 编写（追求高性能执行和严格的内存管理），但提供了深度的 Python 绑定。你作为 20 年经验的程序员，会发现它的 Python API 极度成熟，支持 NumPy、Pandas 和主要的机器学习库。
全球资产全覆盖： 它是极少数在单一引擎中完美支持 股票、期权、期货、外汇、加密货币、CFD 和指数 的框架，且支持跨资产组合交易。
回测与实盘 100% 对等： LEAN 最大的卖点是其“代码即生产”的设计。你在本地回测的代码，可以直接部署到其云端或通过其本地 CLI 接入实盘经纪商（如盈透、币安），逻辑无需改动。

2. 独有的“仿真级”建模能力

LEAN 之所以被很多对冲基金使用，是因为它对市场细节的模拟达到了近乎苛刻的程度：

生存偏误校验（Survivorship Bias Free）： 自动处理股票代码变更、除权除息、分拆合并以及退市数据，确保回测不“作弊”。
纳秒级时间戳： 支持 tick、秒、分钟、小时到天级别的分辨率，能处理极高频的事件触发。
复杂的经纪商模型： 内置了不同券商的费率、滑点、延迟、T+3 结算逻辑以及保证金限制（Margin Calling）模型。

3. 本地与云端的“双栖”开发 (Hybrid Workflow)

这是 LEAN 区别于其他框架的独特商业模式：

LEAN CLI： 你可以完全脱离云端，在自己的服务器上使用 Docker 运行 LEAN。它允许你连接自己的私有数据库（如你提到的 stock_daily_history 表）进行回测。
QuantConnect Cloud： 如果你不想折腾数据，可以把代码上传到其云端。他们拥有超过 400TB 的清洗好的历史数据，直接调用即可。
研究环境（Research Environment）： 集成了 Jupyter Notebook。你可以在 Notebook 中拉取历史数据做统计实验，然后一键将逻辑转为自动化策略。

4. 强大的工具集成

指标库： 内置 100+ 工业级技术指标（不仅是简单的 TA-Lib 封装，更是经过高并发优化的实现）。
宇宙选择（Universe Selection）： 允许你定义复杂的动态股票池逻辑。例如：“每天早上挑选成交量前 10% 且 PE 低于 15 的股票”。
风险管理： 内置风险处理模块，可以自动在策略层面强制执行最大回撤限制、单一头寸上限等。

5. 与同类工具的对比参考

特性	LEAN	NautilusTrader	Backtrader
底层语言	C# (高度成熟)	Rust (极致速度)	Python (易用但慢)
资产支持	最全 (含期权/期货组合)	中等 (侧重数字货币/外汇)	一般
社区规模	巨大 (40万+ 成员)	中等 (极客圈)	大 (但已停止更新)
实盘支持	官方深度集成多券商	需自行开发/配置适配器	较弱

NautilusTrader Chapter 4: Data Import and Processing

Henry Lin — Fri, 06 Feb 2026 02:38:22 +0000

Chapter 4: Data Import and Processing (Part 1)

Learning Objectives

By studying this chapter, you will:

Understand the various data types supported by NautilusTrader
Master methods for importing data from CSV files
Learn to process and clean historical data
Understand the importance of time synchronization
Create and validate high-quality datasets

4.1 Data Types Overview

NautilusTrader supports multiple market data types, each with its specific purpose:

4.1.1 Tick Data

QuoteTick

Contains bid and ask price information, the highest frequency data type:

from nautilus_trader.model.data import QuoteTick
from nautilus_trader.model.objects import Price, Quantity

# Create a quote tick
quote = QuoteTick(
    instrument_id=BTCUSDT_BINANCE,
    bid_price=Price.from_str("50000.00"),
    ask_price=Price.from_str("50001.00"),
    bid_size=Quantity.from_int(10),
    ask_size=Quantity.from_int(15),
    ts_event=1640995200000000000,  # Nanosecond timestamp
    ts_init=1640995200000000000,
)

# Use cases
# - High-frequency trading
# - Order book analysis
# - Latency analysis
# - Slippage calculation

TradeTick

Records actual traded prices and quantities:

from nautilus_trader.model.data import TradeTick
from nautilus_trader.model.enums import AggressorSide

# Create a trade tick
trade = TradeTick(
    instrument_id=BTCUSDT_BINANCE,
    price=Price.from_str("50000.50"),
    size=Quantity.from_int(5),
    aggressor_side=AggressorSide.BUYER,  # Buyer takes the order
    trade_id="123456789",
    ts_event=1640995200000000000,
    ts_init=1640995200000000000,
)

# Use cases
# - Volume analysis
# - Trade flow analysis
# - Microstructure research

4.1.2 Bar Data (K-line)

Bar data is aggregated from tick data, containing OHLCV information:

from nautilus_trader.model.data import Bar, BarType
from nautilus_trader.model.enums import BarAggregation, PriceType

# Define bar type
bar_type = BarType(
    instrument_id=InstrumentId.from_str("BTCUSDT.BINANCE"),
    bar_spec=BarSpecification(
        step=1,
        aggregation=BarAggregation.MINUTE,  # 1-minute bars
        price_type=PriceType.LAST,         # Last price
    ),
    aggregation_source="BINANCE",
)

# Create a bar
bar = Bar(
    bar_type=bar_type,
    open=Price.from_str("50000.00"),
    high=Price.from_str("50100.00"),
    low=Price.from_str("49900.00"),
    close=Price.from_str("50050.00"),
    volume=Quantity.from_int(100),
    ts_event=1640995200000000000,
    ts_init=1640995200000000000,
)

# Use cases
# - Technical analysis
# - Backtesting validation
# - Strategy signals
# - Risk management

4.1.3 Order Book Data

OrderBookSnapshot

from nautilus_trader.model.data import OrderBookSnapshot
from nautilus_trader.model.orderbook import OrderBook
from nautilus_trader.model.enums import OrderSide

# Create order book
book = OrderBook(
    instrument_id=InstrumentId.from_str("BTCUSDT.BINANCE"),
    price_precision=2,
    size_precision=6,
)

# Add orders
book.update(
    OrderSide.BUY,
    Price.from_str("50000.00"),
    Quantity.from_int(10),
    1640995200000000000,
)

book.update(
    OrderSide.SELL,
    Price.from_str("50001.00"),
    Quantity.from_int(15),
    1640995200000000000,
)

# Create snapshot
snapshot = OrderBookSnapshot(
    instrument_id=book.instrument_id,
    bids=book.bids(),
    asks=book.asks(),
    ts_event=1640995200000000000,
    ts_init=1640995200000000000,
)

# Use cases
# - Market making strategies
# - Liquidity analysis
# - Large order detection
# - Slippage prediction

4.2 Importing Data from CSV

CSV is the most common historical data format. Let's create a complete data import workflow.

4.2.1 Preparing CSV Data

Create sample CSV file btcusdt_1m.csv:

timestamp,open,high,low,close,volume
2024-01-01 00:00:00,50000.00,50100.00,49900.00,50050.00,125.5
2024-01-01 00:01:00,50050.00,50200.00,49950.00,50150.00,98.3
2024-01-01 00:02:00,50150.00,50300.00,50000.00,50250.00,156.7
2024-01-01 00:03:00,50250.00,50400.00,50150.00,50350.00,203.4
2024-01-01 00:04:00,50350.00,50500.00,50250.00,50450.00,187.2

4.2.2 Creating a Data Importer

Create data_importer.py:

"""
CSV Data Importer
Supports multiple CSV formats and configuration options
"""

import pandas as pd
from datetime import datetime
from decimal import Decimal
from pathlib import Path
from typing import List, Optional, Dict, Any

from nautilus_trader.model.data import Bar
from nautilus_trader.model.data import BarType
from nautilus_trader.model.data import BarSpecification
from nautilus_trader.model.enums import BarAggregation
from nautilus_trader.model.enums import PriceType
from nautilus_trader.model.identifiers import InstrumentId
from nautilus_trader.persistence.wranglers import BarDataWrangler
from nautilus_trader.test_kit.providers import TestInstrumentProvider


class CSVDataImporter:
    """
    CSV Data Importer

    Supported features:
    - Automatic column name detection
    - Time format parsing
    - Data validation
    - Missing value handling
    """

    def __init__(self):
        """Initialize the importer"""
        # Common column name mappings
        self.column_mappings = {
            'timestamp': ['timestamp', 'datetime', 'time', 'date'],
            'open': ['open', 'o', 'Open'],
            'high': ['high', 'h', 'High'],
            'low': ['low', 'l', 'Low'],
            'close': ['close', 'c', 'Close'],
            'volume': ['volume', 'vol', 'v', 'Volume'],
        }

        # Time format list
        self.time_formats = [
            '%Y-%m-%d %H:%M:%S',
            '%Y-%m-%d %H:%M:%S.%f',
            '%Y-%m-%dT%H:%M:%S',
            '%Y-%m-%dT%H:%M:%S.%f',
            '%Y-%m-%d',
            '%d/%m/%Y %H:%M:%S',
            '%m/%d/%Y %H:%M:%S',
        ]

    def load_csv(self, file_path: Path, **kwargs) -> pd.DataFrame:
        """
        Load CSV file

        Parameters
        ----------
        file_path : Path
            CSV file path

        Returns
        -------
        pd.DataFrame
            Loaded data
        """
        print(f"Loading CSV file: {file_path}")

        # Try different encodings
        encodings = ['utf-8', 'gbk', 'gb2312', 'latin-1']

        for encoding in encodings:
            try:
                # Auto-detect separator
                df = pd.read_csv(
                    file_path,
                    encoding=encoding,
                    **kwargs
                )
                print(f"Successfully loaded using encoding: {encoding}")
                break
            except UnicodeDecodeError:
                continue
        else:
            raise ValueError(f"Unable to decode file: {file_path}")

        # Display basic information
        print(f"Data shape: {df.shape}")
        print(f"Column names: {list(df.columns)}")

        return df

    def normalize_columns(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Normalize column names

        Parameters
        ----------
        df : pd.DataFrame
            Original data

        Returns
        -------
        pd.DataFrame
            Normalized data
        """
        print("Normalizing column names...")

        # Convert to lowercase
        df.columns = df.columns.str.lower()

        # Find and rename columns
        column_map = {}
        for standard_name, possible_names in self.column_mappings.items():
            for col in df.columns:
                if col in possible_names:
                    column_map[col] = standard_name
                    break

        if column_map:
            df = df.rename(columns=column_map)
            print(f"Renamed columns: {column_map}")

        # Check required columns
        required_columns = ['timestamp', 'open', 'high', 'low', 'close']
        missing_columns = [col for col in required_columns if col not in df.columns]

        if missing_columns:
            raise ValueError(f"Missing required columns: {missing_columns}")

        # If no volume column, add default value
        if 'volume' not in df.columns:
            df['volume'] = 1.0
            print("Added default volume column")

        return df

    def parse_timestamp(self, df: pd.DataFrame, timestamp_col: str = 'timestamp') -> pd.DataFrame:
        """
        Parse timestamps

        Parameters
        ----------
        df : pd.DataFrame
            Data containing timestamps
        timestamp_col : str
            Timestamp column name

        Returns
        -------
        pd.DataFrame
            Parsed data
        """
        print("Parsing timestamps...")

        # Try different time formats
        for time_format in self.time_formats:
            try:
                df['timestamp'] = pd.to_datetime(
                    df[timestamp_col],
                    format=time_format
                )
                print(f"Successfully parsed time format: {time_format}")
                break
            except (ValueError, TypeError):
                continue
        else:
            # If all fail, try auto-parsing
            try:
                df['timestamp'] = pd.to_datetime(df[timestamp_col])
                print("Auto-parsed timestamps successfully")
            except Exception as e:
                raise ValueError(f"Unable to parse timestamps: {e}")

        # Check timestamp range
        print(f"Time range: {df['timestamp'].min()} to {df['timestamp'].max()}")

        # Set as index (optional)
        df = df.set_index('timestamp')

        return df

    def validate_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Validate and clean data

        Parameters
        ----------
        df : pd.DataFrame
            Data to validate

        Returns
        -------
        pd.DataFrame
            Validated data
        """
        print("Validating data...")

        # Check numeric columns
        numeric_columns = ['open', 'high', 'low', 'close', 'volume']

        # Check for null values
        null_counts = df[numeric_columns].isnull().sum()
        if null_counts.any():
            print(f"Found null values: {null_counts[null_counts > 0]}")

            # Remove rows containing null values
            df = df.dropna(subset=numeric_columns)
            print("Removed rows containing null values")

        # Convert to numeric types
        for col in numeric_columns:
            df[col] = pd.to_numeric(df[col], errors='coerce')

        # Check price relationships
        invalid_prices = df[
            (df['high'] < df['low']) |
            (df['high'] < df['open']) |
            (df['high'] < df['close']) |
            (df['low'] > df['open']) |
            (df['low'] > df['close'])
        ]

        if len(invalid_prices) > 0:
            print(f"Found {len(invalid_prices)} rows with invalid price relationships")
            # Fix price relationships
            df['high'] = df[['high', 'open', 'close']].max(axis=1)
            df['low'] = df[['low', 'open', 'close']].min(axis=1)
            print("Fixed price relationships")

        # Check negative values
        negative_values = df[
            (df['open'] <= 0) |
            (df['high'] <= 0) |
            (df['low'] <= 0) |
            (df['close'] <= 0)
        ]

        if len(negative_values) > 0:
            print(f"Found {len(negative_values)} rows with negative prices")
            df = df[df['open'] > 0]
            df = df[df['high'] > 0]
            df = df[df['low'] > 0]
            df = df[df['close'] > 0]

        # Check duplicate timestamps
        duplicates = df.index.duplicated().sum()
        if duplicates > 0:
            print(f"Found {duplicates} duplicate timestamps")
            df = df[~df.index.duplicated(keep='first')]

        print(f"Validation complete, remaining data: {len(df)} rows")
        return df

    def convert_to_bars(
        self,
        df: pd.DataFrame,
        instrument_id: InstrumentId,
        bar_type: BarType
    ) -> List[Bar]:
        """
        Convert to Bar objects

        Parameters
        ----------
        df : pd.DataFrame
            Validated data
        instrument_id : InstrumentId
            Instrument ID
        bar_type : BarType
            Bar type

        Returns
        -------
        List[Bar]
            List of Bar objects
        """
        print("Converting to Bar objects...")

        # Create instrument (using test instrument as example)
        instrument = TestInstrumentProvider.btcusdt_binance()

        # Create data wrangler
        wrangler = BarDataWrangler(
            bar_type=bar_type,
            instrument=instrument
        )

        # Reset index to access timestamps
        df = df.reset_index()

        # Convert data
        bars = wrangler.process_df(df)

        print(f"Successfully converted {len(bars)} bars")
        return bars

    def import_csv_to_bars(
        self,
        file_path: Path,
        instrument_id: InstrumentId,
        bar_type: BarType,
        **kwargs
    ) -> List[Bar]:
        """
        Complete workflow for importing from CSV and converting to Bar objects

        Parameters
        ----------
        file_path : Path
            CSV file path
        instrument_id : InstrumentId
            Instrument ID
        bar_type : BarType
            Bar type
        **kwargs
            Parameters to pass to pd.read_csv

        Returns
        -------
        List[Bar]
            List of Bar objects
        """
        # 1. Load CSV
        df = self.load_csv(file_path, **kwargs)

        # 2. Normalize column names
        df = self.normalize_columns(df)

        # 3. Parse timestamps
        df = self.parse_timestamp(df)

        # 4. Validate data
        df = self.validate_data(df)

        # 5. Convert to Bar objects
        bars = self.convert_to_bars(df, instrument_id, bar_type)

        return bars


def main():
    """Main function - Example usage"""

    # Create importer
    importer = CSVDataImporter()

    # Define instrument and bar type
    instrument_id = InstrumentId.from_str("BTCUSDT.BINANCE")
    bar_type = BarType(
        instrument_id=instrument_id,
        bar_spec=BarSpecification(
            step=1,
            aggregation=BarAggregation.MINUTE,
            price_type=PriceType.LAST,
        ),
        aggregation_source="BINANCE",
    )

    # Import data
    csv_path = Path("btcusdt_1m.csv")

    if csv_path.exists():
        try:
            bars = importer.import_csv_to_bars(
                file_path=csv_path,
                instrument_id=instrument_id,
                bar_type=bar_type,
                sep=',',
                skipinitialspace=True,
            )

            # Display results
            print(f"\nSuccessfully imported {len(bars)} bars")
            if bars:
                print(f"First bar: {bars[0]}")
                print(f"Last bar: {bars[-1]}")

        except Exception as e:
            print(f"Import failed: {e}")
    else:
        print(f"File does not exist: {csv_path}")
        print("Please create test CSV file before running")


if __name__ == "__main__":
    main()

4.2.3 Creating a CSV File Generator

For testing, let's create a tool to generate simulated CSV data:

Create generate_csv_data.py:

"""
Generate simulated CSV historical data
"""

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from decimal import Decimal
from pathlib import Path


def generate_price_data(
    days: int = 7,
    start_price: float = 50000.0,
    volatility: float = 0.02,
    trend: float = 0.0001,
    output_path: Path = Path("btcusdt_1m.csv")
):
    """
    Generate simulated 1-minute bar data

    Parameters
    ----------
    days : int
        Number of days to generate
    start_price : float
        Starting price
    volatility : float
        Volatility
    trend : float
        Trend (positive for up, negative for down)
    output_path : Path
        Output file path
    """
    print(f"Generating {days} days of 1-minute bar data...")

    # Calculate number of minutes
    minutes = days * 24 * 60
    timestamps = pd.date_range(
        start=datetime(2024, 1, 1),
        periods=minutes,
        freq='1min'
    )

    # Generate price data
    prices = [start_price]

    for i in range(1, minutes):
        # Random walk + trend
        change = np.random.normal(trend, volatility / np.sqrt(1440))
        new_price = prices[-1] * (1 + change)
        prices.append(max(new_price, 1.0))  # Ensure price is not negative

    prices = np.array(prices)

    # Generate OHLC
    opens = prices[:-1]
    closes = prices[1:]

    # Generate high and low prices (between open and close)
    highs = np.maximum(opens, closes) * (1 + np.random.uniform(0, 0.005, minutes-1))
    lows = np.minimum(opens, closes) * (1 - np.random.uniform(0, 0.005, minutes-1))

    # Generate volume (correlated with price changes)
    volume_base = 100
    volume = volume_base + np.abs(np.random.normal(0, 50, minutes-1))
    volume = np.maximum(volume, 1)

    # Create DataFrame
    df = pd.DataFrame({
        'timestamp': timestamps[:-1],
        'open': opens,
        'high': highs,
        'low': lows,
        'close': closes,
        'volume': volume,
    })

    # Format timestamps
    df['timestamp'] = df['timestamp'].dt.strftime('%Y-%m-%d %H:%M:%S')

    # Save to CSV
    df.to_csv(output_path, index=False)
    print(f"Data saved to: {output_path}")
    print(f"Data range: {df['timestamp'].iloc[0]} to {df['timestamp'].iloc[-1]}")
    print(f"Price range: {df['open'].min():.2f} to {df['high'].max():.2f}")


def main():
    """Main function"""
    # Generate 7 days of data
    generate_price_data(
        days=7,
        start_price=50000.0,
        volatility=0.02,  # 2% daily volatility
        trend=0.0001,     # Slight upward trend
    )


if __name__ == "__main__":
    main()

4.2.4 Running the Example

# 1. Generate test data
python generate_csv_data.py

# 2. Import and process data
python data_importer.py

4.3 Data Quality Control

High-quality data is key to successful backtesting. Here are some important quality control measures:

4.3.1 Data Validation Checklist

Create data_validator.py:

"""
Data validation tool
Ensure data quality
"""

import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
from datetime import datetime


class DataValidator:
    """Data validator"""

    def __init__(self):
        """Initialize validator"""
        self.errors = []
        self.warnings = []

    def validate_completeness(self, df: pd.DataFrame) -> bool:
        """
        Validate data completeness

        Parameters
        ----------
        df : pd.DataFrame
            Data to validate

        Returns
        -------
        bool
            Whether validation passed
        """
        print("Validating data completeness...")

        # Check for missing values
        null_counts = df.isnull().sum()
        if null_counts.any():
            self.errors.append(f"Found missing values: {null_counts[null_counts > 0].to_dict()}")
            return False

        # Check empty dataset
        if len(df) == 0:
            self.errors.append("Dataset is empty")
            return False

        print(f"Data completeness validation passed, total {len(df)} rows")
        return True

    def validate_time_sequence(self, df: pd.DataFrame, timestamp_col: str = 'timestamp') -> bool:
        """
        Validate time sequence

        Parameters
        ----------
        df : pd.DataFrame
            Data to validate
        timestamp_col : str
            Timestamp column name

        Returns
        -------
        bool
            Whether validation passed
        """
        print("Validating time sequence...")

        if timestamp_col not in df.columns:
            self.errors.append(f"Missing timestamp column: {timestamp_col}")
            return False

        # Ensure timestamp is datetime type
        if not pd.api.types.is_datetime64_any_dtype(df[timestamp_col]):
            df[timestamp_col] = pd.to_datetime(df[timestamp_col])

        # Check if timestamps are increasing
        time_diff = df[timestamp_col].diff()
        if (time_diff <= 0).any():
            invalid_count = (time_diff <= 0).sum()
            self.errors.append(f"Found {invalid_count} non-increasing timestamps")
            return False

        # Check time intervals
        min_interval = time_diff.min()
        max_interval = time_diff.max()

        print(f"Time interval range: {min_interval} to {max_interval}")

        # Check for abnormal time intervals
        expected_interval = pd.Timedelta(minutes=1)  # Assuming 1-minute data
        tolerance = pd.Timedelta(seconds=30)

        irregular_intervals = df[time_diff > expected_interval + tolerance]
        if len(irregular_intervals) > 0:
            self.warnings.append(f"Found {len(irregular_intervals)} abnormal time intervals")

        print("Time sequence validation passed")
        return True

    def validate_price_relationships(self, df: pd.DataFrame) -> bool:
        """
        Validate price relationships

        Parameters
        ----------
        df : pd.DataFrame
            Data to validate

        Returns
        -------
        bool
            Whether validation passed
        """
        print("Validating price relationships...")

        required_columns = ['open', 'high', 'low', 'close']
        missing_columns = [col for col in required_columns if col not in df.columns]

        if missing_columns:
            self.errors.append(f"Missing price columns: {missing_columns}")
            return False

        # Check price relationships
        invalid_high = df['high'] < df[['open', 'close']].max(axis=1)
        invalid_low = df['low'] > df[['open', 'close']].min(axis=1)

        total_invalid = (invalid_high | invalid_low).sum()

        if total_invalid > 0:
            self.errors.append(f"Found {total_invalid} rows with invalid price relationships")
            return False

        # Check if prices are positive
        negative_prices = (
            (df['open'] <= 0) |
            (df['high'] <= 0) |
            (df['low'] <= 0) |
            (df['close'] <= 0)
        ).sum()

        if negative_prices > 0:
            self.errors.append(f"Found {negative_prices} rows with negative or zero prices")
            return False

        print("Price relationships validation passed")
        return True

    def validate_volume(self, df: pd.DataFrame) -> bool:
        """
        Validate volume

        Parameters
        ----------
        df : pd.DataFrame
            Data to validate

        Returns
        -------
        bool
            Whether validation passed
        """
        print("Validating volume...")

        if 'volume' not in df.columns:
            self.warnings.append("Missing volume column")
            return True

        # Check for negative volume
        negative_volume = (df['volume'] < 0).sum()
        if negative_volume > 0:
            self.errors.append(f"Found {negative_volume} rows with negative volume")
            return False

        # Check for zero volume
        zero_volume = (df['volume'] == 0).sum()
        if zero_volume > 0:
            self.warnings.append(f"Found {zero_volume} rows with zero volume")

        # Check for abnormal volume (3 times the mean)
        volume_mean = df['volume'].mean()
        volume_std = df['volume'].std()
        threshold = volume_mean + 3 * volume_std

        outliers = df['volume'] > threshold
        outlier_count = outliers.sum()

        if outlier_count > 0:
            self.warnings.append(
                f"Found {outlier_count} rows with abnormal volume "
                f"(>{threshold:.2f}, mean: {volume_mean:.2f})"
            )

        print("Volume validation passed")
        return True

    def validate_duplicates(self, df: pd.DataFrame) -> bool:
        """
        Validate duplicate data

        Parameters
        ----------
        df : pd.DataFrame
            Data to validate

        Returns
        -------
        bool
            Whether validation passed
        """
        print("Validating duplicate data...")

        # Check completely duplicate rows
        duplicate_rows = df.duplicated().sum()
        if duplicate_rows > 0:
            self.warnings.append(f"Found {duplicate_rows} completely duplicate rows")

        # Check duplicate timestamps if timestamp column exists
        if 'timestamp' in df.columns:
            duplicate_timestamps = df['timestamp'].duplicated().sum()
            if duplicate_timestamps > 0:
                self.errors.append(f"Found {duplicate_timestamps} duplicate timestamps")
                return False

        print("Duplicate data validation passed")
        return True

    def generate_report(self) -> Dict:
        """
        Generate validation report

        Returns
        -------
        Dict
            Validation report
        """
        report = {
            'validation_time': datetime.now(),
            'total_errors': len(self.errors),
            'total_warnings': len(self.warnings),
            'errors': self.errors,
            'warnings': self.warnings,
            'status': 'PASSED' if len(self.errors) == 0 else 'FAILED'
        }

        return report

    def validate_all(self, df: pd.DataFrame) -> Dict:
        """
        Execute all validations

        Parameters
        ----------
        df : pd.DataFrame
            Data to validate

        Returns
        -------
        Dict
            Validation report
        """
        print("\nStarting data validation...")
        print("=" * 50)

        # Execute all validations
        validations = [
            self.validate_completeness,
            self.validate_time_sequence,
            self.validate_price_relationships,
            self.validate_volume,
            self.validate_duplicates,
        ]

        for validation in validations:
            validation(df)

        # Generate report
        report = self.generate_report()

        # Print report
        print("\n" + "=" * 50)
        print("Validation Report")
        print("=" * 50)
        print(f"Status: {report['status']}")
        print(f"Error count: {report['total_errors']}")
        print(f"Warning count: {report['total_warnings']}")

        if report['errors']:
            print("\nErrors:")
            for error in report['errors']:
                print(f"  - {error}")

        if report['warnings']:
            print("\nWarnings:")
            for warning in report['warnings']:
                print(f"  - {warning}")

        return report


def main():
    """Main function - Example usage"""

    # Create validator
    validator = DataValidator()

    # Load data
    csv_path = Path("btcusdt_1m.csv")
    if csv_path.exists():
        df = pd.read_csv(csv_path)
        df['timestamp'] = pd.to_datetime(df['timestamp'])

        # Validate data
        report = validator.validate_all(df)

        # Save report
        report_path = Path("validation_report.json")
        import json
        with open(report_path, 'w') as f:
            # Convert datetime object to string
            report['validation_time'] = str(report['validation_time'])
            json.dump(report, f, indent=2)

        print(f"\nValidation report saved to: {report_path}")
    else:
        print(f"File does not exist: {csv_path}")


if __name__ == "__main__":
    main()

4.3.2 Data Cleaning Strategies

For identified issues, we need corresponding cleaning strategies:

"""
Data cleaning tool
"""
import pandas as pd
import numpy as np
from typing import Optional


def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    General function for cleaning data

    Parameters
    ----------
    df : pd.DataFrame
        Original data

    Returns
    -------
    pd.DataFrame
        Cleaned data
    """
    print("Starting data cleaning...")
    original_rows = len(df)

    # 1. Remove completely duplicate rows
    df = df.drop_duplicates()

    # 2. Handle duplicate timestamps (keep first)
    if 'timestamp' in df.columns:
        df = df.drop_duplicates(subset=['timestamp'], keep='first')

    # 3. Handle missing values
    numeric_columns = df.select_dtypes(include=[np.number]).columns
    df[numeric_columns] = df[numeric_columns].fillna(method='ffill').fillna(method='bfill')

    # 4. Fix price relationships
    if all(col in df.columns for col in ['open', 'high', 'low', 'close']):
        # Ensure high is the maximum
        df['high'] = df[['high', 'open', 'close']].max(axis=1)
        # Ensure low is the minimum
        df['low'] = df[['low', 'open', 'close']].min(axis=1)

    # 5. Handle outliers
    for col in ['open', 'high', 'low', 'close']:
        if col in df.columns:
            # Use IQR method to detect outliers
            Q1 = df[col].quantile(0.25)
            Q3 = df[col].quantile(0.75)
            IQR = Q3 - Q1

            # Define outlier range
            lower_bound = Q1 - 1.5 * IQR
            upper_bound = Q3 + 1.5 * IQR

            # Clip outliers
            df[col] = df[col].clip(lower=lower_bound, upper_bound=upper_bound)

    # 6. Ensure prices are positive
    price_columns = ['open', 'high', 'low', 'close']
    for col in price_columns:
        if col in df.columns:
            df[col] = df[col].abs()

    cleaned_rows = len(df)
    print(f"Data cleaning complete: {original_rows} -> {cleaned_rows} rows")

    return df

4.4 Time Synchronization Processing

In real trading, data comes from multiple sources, making time synchronization crucial.

4.4.1 Time Zone Handling

"""
Time synchronization tool
Handles different time zones and timestamp formats
"""

import pandas as pd
from datetime import datetime, timezone
import pytz


class TimeSynchronizer:
    """Time synchronizer"""

    def __init__(self, target_timezone: str = 'UTC'):
        """
        Initialize time synchronizer

        Parameters
        ----------
        target_timezone : str
            Target timezone
        """
        self.target_tz = pytz.timezone(target_timezone)

    def convert_timezone(
        self,
        df: pd.DataFrame,
        timestamp_col: str,
        source_timezone: str
    ) -> pd.DataFrame:
        """
        Convert time zone

        Parameters
        ----------
        df : pd.DataFrame
            Data
        timestamp_col : str
            Timestamp column name
        source_timezone : str
            Source timezone

        Returns
        -------
        pd.DataFrame
            Converted data
        """
        print(f"Converting timezone: {source_timezone} -> {self.target_tz}")

        source_tz = pytz.timezone(source_timezone)

        # Localize timestamps
        df[timestamp_col] = df[timestamp_col].dt.tz_localize(source_tz)

        # Convert to target timezone
        df[timestamp_col] = df[timestamp_col].dt.tz_convert(self.target_tz)

        return df

    def align_time_series(
        self,
        df_list: list[pd.DataFrame],
        timestamp_col: str = 'timestamp'
    ) -> list[pd.DataFrame]:
        """
        Align multiple time series

        Parameters
        ----------
        df_list : list[pd.DataFrame]
            List of dataframes
        timestamp_col : str
            Timestamp column name

        Returns
        -------
        list[pd.DataFrame]
            List of aligned dataframes
        """
        print("Aligning time series...")

        # Find intersection of all timestamps
        all_timestamps = None

        for df in df_list:
            timestamps = set(df[timestamp_col])
            if all_timestamps is None:
                all_timestamps = timestamps
            else:
                all_timestamps &= timestamps

        print(f"Common timestamp count: {len(all_timestamps)}")

        # Filter each dataframe
        aligned_dfs = []
        for df in df_list:
            aligned_df = df[df[timestamp_col].isin(all_timestamps)]
            aligned_dfs.append(aligned_df.sort_values(timestamp_col))

        return aligned_dfs

4.5 Next Steps

In the first part of this chapter, we learned:

Data types supported by NautilusTrader
Complete workflow for importing data from CSV
Data quality control methods
Time synchronization processing

In the next part, we will learn:

Using Parquet data directories
Efficient data storage and retrieval
Data vendor integration
Real-time data processing

4.6 Summary

Key Points

NautilusTrader supports multiple market data types
CSV import requires careful data validation and cleaning
Time synchronization is crucial for multi-source data
High-quality data is the foundation for successful backtesting

Practical Recommendations

Always validate imported data
Save data validation reports
Use appropriate data cleaning strategies
Consider using Parquet format for improved performance

4.7 References

NautilusTrader 第4章：数据导入与处理

Henry Lin — Fri, 06 Feb 2026 02:37:40 +0000

第4章：数据导入与处理

学习目标

通过本章学习，您将：

了解 NautilusTrader 支持的各种数据类型
掌握从 CSV 文件导入数据的方法
学会处理和清洗历史数据
理解时间同步的重要性
创建和验证高质量的数据集

4.1 数据类型概览

NautilusTrader 支持多种市场数据类型，每种数据都有其特定的用途：

4.1.1 Tick 数据

QuoteTick（报价数据）

包含买卖价格信息，是最高频的数据类型：

from nautilus_trader.model.data import QuoteTick
from nautilus_trader.model.objects import Price, Quantity

# 创建报价tick
quote = QuoteTick(
    instrument_id=BTCUSDT_BINANCE,
    bid_price=Price.from_str("50000.00"),
    ask_price=Price.from_str("50001.00"),
    bid_size=Quantity.from_int(10),
    ask_size=Quantity.from_int(15),
    ts_event=1640995200000000000,  # 纳秒时间戳
    ts_init=1640995200000000000,
)

# 使用场景
# - 高频交易
# - 订单簿分析
# - 延迟分析
# - 滑点计算

TradeTick（成交数据）

记录实际成交的价格和数量：

from nautilus_trader.model.data import TradeTick
from nautilus_trader.model.enums import AggressorSide

# 创建成交tick
trade = TradeTick(
    instrument_id=BTCUSDT_BINANCE,
    price=Price.from_str("50000.50"),
    size=Quantity.from_int(5),
    aggressor_side=AggressorSide.BUYER,  # 买方吃单
    trade_id="123456789",
    ts_event=1640995200000000000,
    ts_init=1640995200000000000,
)

# 使用场景
# - 成交量分析
# - 交易流向分析
# - 微观结构研究

4.1.2 K线数据（Bar）

K线数据是tick数据的聚合，包含OHLCV信息：

from nautilus_trader.model.data import Bar, BarType
from nautilus_trader.model.enums import BarAggregation, PriceType

# 定义K线类型
bar_type = BarType(
    instrument_id=InstrumentId.from_str("BTCUSDT.BINANCE"),
    bar_spec=BarSpecification(
        step=1,
        aggregation=BarAggregation.MINUTE,  # 1分钟K线
        price_type=PriceType.LAST,         # 成交价
    ),
    aggregation_source="BINANCE",
)

# 创建K线
bar = Bar(
    bar_type=bar_type,
    open=Price.from_str("50000.00"),
    high=Price.from_str("50100.00"),
    low=Price.from_str("49900.00"),
    close=Price.from_str("50050.00"),
    volume=Quantity.from_int(100),
    ts_event=1640995200000000000,
    ts_init=1640995200000000000,
)

# 使用场景
# - 技术分析
# - 回测验证
# - 策略信号
# - 风险管理

4.1.3 订单簿数据

OrderBookSnapshot（订单簿快照）

from nautilus_trader.model.data import OrderBookSnapshot
from nautilus_trader.model.orderbook import OrderBook
from nautilus_trader.model.enums import OrderSide

# 创建订单簿
book = OrderBook(
    instrument_id=InstrumentId.from_str("BTCUSDT.BINANCE"),
    price_precision=2,
    size_precision=6,
)

# 添加订单
book.update(
    OrderSide.BUY,
    Price.from_str("50000.00"),
    Quantity.from_int(10),
    1640995200000000000,
)

book.update(
    OrderSide.SELL,
    Price.from_str("50001.00"),
    Quantity.from_int(15),
    1640995200000000000,
)

# 创建快照
snapshot = OrderBookSnapshot(
    instrument_id=book.instrument_id,
    bids=book.bids(),
    asks=book.asks(),
    ts_event=1640995200000000000,
    ts_init=1640995200000000000,
)

# 使用场景
# - 做市策略
# - 流动性分析
# - 大单检测
# - 滑点预测

4.2 从 CSV 导入数据

CSV 是最常见的历史数据格式。让我们创建一个完整的数据导入流程。

4.2.1 准备 CSV 数据

创建示例 CSV 文件 btcusdt_1m.csv：

timestamp,open,high,low,close,volume
2024-01-01 00:00:00,50000.00,50100.00,49900.00,50050.00,125.5
2024-01-01 00:01:00,50050.00,50200.00,49950.00,50150.00,98.3
2024-01-01 00:02:00,50150.00,50300.00,50000.00,50250.00,156.7
2024-01-01 00:03:00,50250.00,50400.00,50150.00,50350.00,203.4
2024-01-01 00:04:00,50350.00,50500.00,50250.00,50450.00,187.2

4.2.2 创建数据导入器

创建 data_importer.py：

"""
CSV 数据导入器
支持多种 CSV 格式和配置选项
"""

import pandas as pd
from datetime import datetime
from decimal import Decimal
from pathlib import Path
from typing import List, Optional, Dict, Any

from nautilus_trader.model.data import Bar
from nautilus_trader.model.data import BarType
from nautilus_trader.model.data import BarSpecification
from nautilus_trader.model.enums import BarAggregation
from nautilus_trader.model.enums import PriceType
from nautilus_trader.model.identifiers import InstrumentId
from nautilus_trader.persistence.wranglers import BarDataWrangler
from nautilus_trader.test_kit.providers import TestInstrumentProvider


class CSVDataImporter:
    """
    CSV 数据导入器

    支持的功能：
    - 自动检测列名
    - 时间格式解析
    - 数据验证
    - 缺失值处理
    """

    def __init__(self):
        """初始化导入器"""
        # 常见的列名映射
        self.column_mappings = {
            'timestamp': ['timestamp', 'datetime', 'time', 'date'],
            'open': ['open', 'o', 'Open'],
            'high': ['high', 'h', 'High'],
            'low': ['low', 'l', 'Low'],
            'close': ['close', 'c', 'Close'],
            'volume': ['volume', 'vol', 'v', 'Volume'],
        }

        # 时间格式列表
        self.time_formats = [
            '%Y-%m-%d %H:%M:%S',
            '%Y-%m-%d %H:%M:%S.%f',
            '%Y-%m-%dT%H:%M:%S',
            '%Y-%m-%dT%H:%M:%S.%f',
            '%Y-%m-%d',
            '%d/%m/%Y %H:%M:%S',
            '%m/%d/%Y %H:%M:%S',
        ]

    def load_csv(self, file_path: Path, **kwargs) -> pd.DataFrame:
        """
        加载 CSV 文件

        Parameters
        ----------
        file_path : Path
            CSV 文件路径

        Returns
        -------
        pd.DataFrame
            加载的数据
        """
        print(f"加载 CSV 文件: {file_path}")

        # 尝试不同的编码
        encodings = ['utf-8', 'gbk', 'gb2312', 'latin-1']

        for encoding in encodings:
            try:
                # 自动检测分隔符
                df = pd.read_csv(
                    file_path,
                    encoding=encoding,
                    **kwargs
                )
                print(f"成功加载，使用编码: {encoding}")
                break
            except UnicodeDecodeError:
                continue
        else:
            raise ValueError(f"无法解码文件: {file_path}")

        # 显示基本信息
        print(f"数据形状: {df.shape}")
        print(f"列名: {list(df.columns)}")

        return df

    def normalize_columns(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        标准化列名

        Parameters
        ----------
        df : pd.DataFrame
            原始数据

        Returns
        -------
        pd.DataFrame
            标准化后的数据
        """
        print("标准化列名...")

        # 转换为小写
        df.columns = df.columns.str.lower()

        # 查找并重命名列
        column_map = {}
        for standard_name, possible_names in self.column_mappings.items():
            for col in df.columns:
                if col in possible_names:
                    column_map[col] = standard_name
                    break

        if column_map:
            df = df.rename(columns=column_map)
            print(f"重命名列: {column_map}")

        # 检查必需的列
        required_columns = ['timestamp', 'open', 'high', 'low', 'close']
        missing_columns = [col for col in required_columns if col not in df.columns]

        if missing_columns:
            raise ValueError(f"缺少必需的列: {missing_columns}")

        # 如果没有volume列，添加默认值
        if 'volume' not in df.columns:
            df['volume'] = 1.0
            print("添加默认volume列")

        return df

    def parse_timestamp(self, df: pd.DataFrame, timestamp_col: str = 'timestamp') -> pd.DataFrame:
        """
        解析时间戳

        Parameters
        ----------
        df : pd.DataFrame
            包含时间戳的数据
        timestamp_col : str
            时间戳列名

        Returns
        -------
        pd.DataFrame
            解析后的数据
        """
        print("解析时间戳...")

        # 尝试不同的时间格式
        for time_format in self.time_formats:
            try:
                df['timestamp'] = pd.to_datetime(
                    df[timestamp_col],
                    format=time_format
                )
                print(f"成功解析时间格式: {time_format}")
                break
            except (ValueError, TypeError):
                continue
        else:
            # 如果都失败了，尝试自动解析
            try:
                df['timestamp'] = pd.to_datetime(df[timestamp_col])
                print("自动解析时间戳成功")
            except Exception as e:
                raise ValueError(f"无法解析时间戳: {e}")

        # 检查时间戳范围
        print(f"时间范围: {df['timestamp'].min()} 到 {df['timestamp'].max()}")

        # 设置为索引（可选）
        df = df.set_index('timestamp')

        return df

    def validate_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        验证和清洗数据

        Parameters
        ----------
        df : pd.DataFrame
            待验证的数据

        Returns
        -------
        pd.DataFrame
            验证后的数据
        """
        print("验证数据...")

        # 检查数值列
        numeric_columns = ['open', 'high', 'low', 'close', 'volume']

        # 检查是否有空值
        null_counts = df[numeric_columns].isnull().sum()
        if null_counts.any():
            print(f"发现空值: {null_counts[null_counts > 0]}")

            # 删除包含空值的行
            df = df.dropna(subset=numeric_columns)
            print("删除包含空值的行")

        # 转换为数值类型
        for col in numeric_columns:
            df[col] = pd.to_numeric(df[col], errors='coerce')

        # 检查价格关系
        invalid_prices = df[
            (df['high'] < df['low']) |
            (df['high'] < df['open']) |
            (df['high'] < df['close']) |
            (df['low'] > df['open']) |
            (df['low'] > df['close'])
        ]

        if len(invalid_prices) > 0:
            print(f"发现 {len(invalid_prices)} 行无效的价格关系")
            # 修正价格关系
            df['high'] = df[['high', 'open', 'close']].max(axis=1)
            df['low'] = df[['low', 'open', 'close']].min(axis=1)
            print("修正价格关系")

        # 检查负值
        negative_values = df[
            (df['open'] <= 0) |
            (df['high'] <= 0) |
            (df['low'] <= 0) |
            (df['close'] <= 0)
        ]

        if len(negative_values) > 0:
            print(f"发现 {len(negative_values)} 行负价格")
            df = df[df['open'] > 0]
            df = df[df['high'] > 0]
            df = df[df['low'] > 0]
            df = df[df['close'] > 0]

        # 检查重复时间戳
        duplicates = df.index.duplicated().sum()
        if duplicates > 0:
            print(f"发现 {duplicates} 个重复时间戳")
            df = df[~df.index.duplicated(keep='first')]

        print(f"验证完成，剩余数据: {len(df)} 行")
        return df

    def convert_to_bars(
        self,
        df: pd.DataFrame,
        instrument_id: InstrumentId,
        bar_type: BarType
    ) -> List[Bar]:
        """
        转换为 Bar 对象

        Parameters
        ----------
        df : pd.DataFrame
            验证后的数据
        instrument_id : InstrumentId
            交易工具ID
        bar_type : BarType
            K线类型

        Returns
        -------
        List[Bar]
            Bar对象列表
        """
        print("转换为 Bar 对象...")

        # 创建交易工具（使用测试工具作为示例）
        instrument = TestInstrumentProvider.btcusdt_binance()

        # 创建数据整理器
        wrangler = BarDataWrangler(
            bar_type=bar_type,
            instrument=instrument
        )

        # 重置索引以便访问时间戳
        df = df.reset_index()

        # 转换数据
        bars = wrangler.process_df(df)

        print(f"成功转换 {len(bars)} 根K线")
        return bars

    def import_csv_to_bars(
        self,
        file_path: Path,
        instrument_id: InstrumentId,
        bar_type: BarType,
        **kwargs
    ) -> List[Bar]:
        """
        从CSV导入并转换为Bar对象的完整流程

        Parameters
        ----------
        file_path : Path
            CSV文件路径
        instrument_id : InstrumentId
            交易工具ID
        bar_type : BarType
            K线类型
        **kwargs
            传递给pd.read_csv的参数

        Returns
        -------
        List[Bar]
            Bar对象列表
        """
        # 1. 加载CSV
        df = self.load_csv(file_path, **kwargs)

        # 2. 标准化列名
        df = self.normalize_columns(df)

        # 3. 解析时间戳
        df = self.parse_timestamp(df)

        # 4. 验证数据
        df = self.validate_data(df)

        # 5. 转换为Bar对象
        bars = self.convert_to_bars(df, instrument_id, bar_type)

        return bars


def main():
    """主函数 - 示例用法"""

    # 创建导入器
    importer = CSVDataImporter()

    # 定义交易工具和K线类型
    instrument_id = InstrumentId.from_str("BTCUSDT.BINANCE")
    bar_type = BarType(
        instrument_id=instrument_id,
        bar_spec=BarSpecification(
            step=1,
            aggregation=BarAggregation.MINUTE,
            price_type=PriceType.LAST,
        ),
        aggregation_source="BINANCE",
    )

    # 导入数据
    csv_path = Path("btcusdt_1m.csv")

    if csv_path.exists():
        try:
            bars = importer.import_csv_to_bars(
                file_path=csv_path,
                instrument_id=instrument_id,
                bar_type=bar_type,
                sep=',',
                skipinitialspace=True,
            )

            # 显示结果
            print(f"\n成功导入 {len(bars)} 根K线")
            if bars:
                print(f"第一根K线: {bars[0]}")
                print(f"最后一根K线: {bars[-1]}")

        except Exception as e:
            print(f"导入失败: {e}")
    else:
        print(f"文件不存在: {csv_path}")
        print("请创建测试CSV文件后再运行")


if __name__ == "__main__":
    main()

4.2.3 创建CSV文件生成器

为了测试，我们创建一个生成模拟CSV数据的工具：

创建 generate_csv_data.py：

"""
生成模拟的CSV历史数据
"""

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from decimal import Decimal
from pathlib import Path


def generate_price_data(
    days: int = 7,
    start_price: float = 50000.0,
    volatility: float = 0.02,
    trend: float = 0.0001,
    output_path: Path = Path("btcusdt_1m.csv")
):
    """
    生成模拟的1分钟K线数据

    Parameters
    ----------
    days : int
        生成天数
    start_price : float
        起始价格
    volatility : float
        波动率
    trend : float
        趋势（正值上涨，负值下跌）
    output_path : Path
        输出文件路径
    """
    print(f"生成 {days} 天的1分钟K线数据...")

    # 计算分钟数
    minutes = days * 24 * 60
    timestamps = pd.date_range(
        start=datetime(2024, 1, 1),
        periods=minutes,
        freq='1min'
    )

    # 生成价格数据
    prices = [start_price]

    for i in range(1, minutes):
        # 随机游走 + 趋势
        change = np.random.normal(trend, volatility / np.sqrt(1440))
        new_price = prices[-1] * (1 + change)
        prices.append(max(new_price, 1.0))  # 确保价格不为负

    prices = np.array(prices)

    # 生成OHLC
    opens = prices[:-1]
    closes = prices[1:]

    # 生成高低价（在开盘和收盘之间）
    highs = np.maximum(opens, closes) * (1 + np.random.uniform(0, 0.005, minutes-1))
    lows = np.minimum(opens, closes) * (1 - np.random.uniform(0, 0.005, minutes-1))

    # 生成成交量（与价格变动相关）
    volume_base = 100
    volume = volume_base + np.abs(np.random.normal(0, 50, minutes-1))
    volume = np.maximum(volume, 1)

    # 创建DataFrame
    df = pd.DataFrame({
        'timestamp': timestamps[:-1],
        'open': opens,
        'high': highs,
        'low': lows,
        'close': closes,
        'volume': volume,
    })

    # 格式化时间戳
    df['timestamp'] = df['timestamp'].dt.strftime('%Y-%m-%d %H:%M:%S')

    # 保存到CSV
    df.to_csv(output_path, index=False)
    print(f"数据已保存到: {output_path}")
    print(f"数据范围: {df['timestamp'].iloc[0]} 到 {df['timestamp'].iloc[-1]}")
    print(f"价格范围: {df['open'].min():.2f} 到 {df['high'].max():.2f}")


def main():
    """主函数"""
    # 生成7天的数据
    generate_price_data(
        days=7,
        start_price=50000.0,
        volatility=0.02,  # 2% 日波动率
        trend=0.0001,     # 轻微上涨趋势
    )


if __name__ == "__main__":
    main()

4.2.4 运行示例

# 1. 生成测试数据
python generate_csv_data.py

# 2. 导入并处理数据
python data_importer.py

4.3 数据质量控制

高质量的数据是回测成功的关键。以下是一些重要的质量控制措施：

4.3.1 数据验证清单

创建 data_validator.py：

"""
数据验证工具
确保数据质量
"""

import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
from datetime import datetime


class DataValidator:
    """数据验证器"""

    def __init__(self):
        """初始化验证器"""
        self.errors = []
        self.warnings = []

    def validate_completeness(self, df: pd.DataFrame) -> bool:
        """
        验证数据完整性

        Parameters
        ----------
        df : pd.DataFrame
            待验证的数据

        Returns
        -------
        bool
            验证是否通过
        """
        print("验证数据完整性...")

        # 检查缺失值
        null_counts = df.isnull().sum()
        if null_counts.any():
            self.errors.append(f"发现缺失值: {null_counts[null_counts > 0].to_dict()}")
            return False

        # 检查空数据集
        if len(df) == 0:
            self.errors.append("数据集为空")
            return False

        print(f"数据完整性验证通过，共 {len(df)} 行")
        return True

    def validate_time_sequence(self, df: pd.DataFrame, timestamp_col: str = 'timestamp') -> bool:
        """
        验证时间序列

        Parameters
        ----------
        df : pd.DataFrame
            待验证的数据
        timestamp_col : str
            时间戳列名

        Returns
        -------
        bool
            验证是否通过
        """
        print("验证时间序列...")

        if timestamp_col not in df.columns:
            self.errors.append(f"缺少时间戳列: {timestamp_col}")
            return False

        # 确保时间戳是datetime类型
        if not pd.api.types.is_datetime64_any_dtype(df[timestamp_col]):
            df[timestamp_col] = pd.to_datetime(df[timestamp_col])

        # 检查时间戳是否递增
        time_diff = df[timestamp_col].diff()
        if (time_diff <= 0).any():
            invalid_count = (time_diff <= 0).sum()
            self.errors.append(f"发现 {invalid_count} 个非递增的时间戳")
            return False

        # 检查时间间隔
        min_interval = time_diff.min()
        max_interval = time_diff.max()

        print(f"时间间隔范围: {min_interval} 到 {max_interval}")

        # 检查是否有异常的时间间隔
        expected_interval = pd.Timedelta(minutes=1)  # 假设是1分钟数据
        tolerance = pd.Timedelta(seconds=30)

        irregular_intervals = df[time_diff > expected_interval + tolerance]
        if len(irregular_intervals) > 0:
            self.warnings.append(f"发现 {len(irregular_intervals)} 个异常时间间隔")

        print("时间序列验证通过")
        return True

    def validate_price_relationships(self, df: pd.DataFrame) -> bool:
        """
        验证价格关系

        Parameters
        ----------
        df : pd.DataFrame
            待验证的数据

        Returns
        -------
        bool
            验证是否通过
        """
        print("验证价格关系...")

        required_columns = ['open', 'high', 'low', 'close']
        missing_columns = [col for col in required_columns if col not in df.columns]

        if missing_columns:
            self.errors.append(f"缺少价格列: {missing_columns}")
            return False

        # 检查价格关系
        invalid_high = df['high'] < df[['open', 'close']].max(axis=1)
        invalid_low = df['low'] > df[['open', 'close']].min(axis=1)

        total_invalid = (invalid_high | invalid_low).sum()

        if total_invalid > 0:
            self.errors.append(f"发现 {total_invalid} 行无效的价格关系")
            return False

        # 检查价格是否为正数
        negative_prices = (
            (df['open'] <= 0) |
            (df['high'] <= 0) |
            (df['low'] <= 0) |
            (df['close'] <= 0)
        ).sum()

        if negative_prices > 0:
            self.errors.append(f"发现 {negative_prices} 行负价格或零价格")
            return False

        print("价格关系验证通过")
        return True

    def validate_volume(self, df: pd.DataFrame) -> bool:
        """
        验证成交量

        Parameters
        ----------
        df : pd.DataFrame
            待验证的数据

        Returns
        -------
        bool
            验证是否通过
        """
        print("验证成交量...")

        if 'volume' not in df.columns:
            self.warnings.append("缺少volume列")
            return True

        # 检查负成交量
        negative_volume = (df['volume'] < 0).sum()
        if negative_volume > 0:
            self.errors.append(f"发现 {negative_volume} 行负成交量")
            return False

        # 检查零成交量
        zero_volume = (df['volume'] == 0).sum()
        if zero_volume > 0:
            self.warnings.append(f"发现 {zero_volume} 行零成交量")

        # 检查异常成交量（均值的三倍）
        volume_mean = df['volume'].mean()
        volume_std = df['volume'].std()
        threshold = volume_mean + 3 * volume_std

        outliers = df['volume'] > threshold
        outlier_count = outliers.sum()

        if outlier_count > 0:
            self.warnings.append(
                f"发现 {outlier_count} 行异常成交量 "
                f"(>{threshold:.2f}, 均值: {volume_mean:.2f})"
            )

        print("成交量验证通过")
        return True

    def validate_duplicates(self, df: pd.DataFrame) -> bool:
        """
        验证重复数据

        Parameters
        ----------
        df : pd.DataFrame
            待验证的数据

        Returns
        -------
        bool
            验证是否通过
        """
        print("验证重复数据...")

        # 检查完全重复的行
        duplicate_rows = df.duplicated().sum()
        if duplicate_rows > 0:
            self.warnings.append(f"发现 {duplicate_rows} 行完全重复的数据")

        # 如果有时间戳列，检查重复的时间戳
        if 'timestamp' in df.columns:
            duplicate_timestamps = df['timestamp'].duplicated().sum()
            if duplicate_timestamps > 0:
                self.errors.append(f"发现 {duplicate_timestamps} 个重复的时间戳")
                return False

        print("重复数据验证通过")
        return True

    def generate_report(self) -> Dict:
        """
        生成验证报告

        Returns
        -------
        Dict
            验证报告
        """
        report = {
            'validation_time': datetime.now(),
            'total_errors': len(self.errors),
            'total_warnings': len(self.warnings),
            'errors': self.errors,
            'warnings': self.warnings,
            'status': 'PASSED' if len(self.errors) == 0 else 'FAILED'
        }

        return report

    def validate_all(self, df: pd.DataFrame) -> Dict:
        """
        执行所有验证

        Parameters
        ----------
        df : pd.DataFrame
            待验证的数据

        Returns
        -------
        Dict
            验证报告
        """
        print("\n开始数据验证...")
        print("=" * 50)

        # 执行所有验证
        validations = [
            self.validate_completeness,
            self.validate_time_sequence,
            self.validate_price_relationships,
            self.validate_volume,
            self.validate_duplicates,
        ]

        for validation in validations:
            validation(df)

        # 生成报告
        report = self.generate_report()

        # 打印报告
        print("\n" + "=" * 50)
        print("验证报告")
        print("=" * 50)
        print(f"状态: {report['status']}")
        print(f"错误数: {report['total_errors']}")
        print(f"警告数: {report['total_warnings']}")

        if report['errors']:
            print("\n错误:")
            for error in report['errors']:
                print(f"  - {error}")

        if report['warnings']:
            print("\n警告:")
            for warning in report['warnings']:
                print(f"  - {warning}")

        return report


def main():
    """主函数 - 示例用法"""

    # 创建验证器
    validator = DataValidator()

    # 加载数据
    csv_path = Path("btcusdt_1m.csv")
    if csv_path.exists():
        df = pd.read_csv(csv_path)
        df['timestamp'] = pd.to_datetime(df['timestamp'])

        # 验证数据
        report = validator.validate_all(df)

        # 保存报告
        report_path = Path("validation_report.json")
        import json
        with open(report_path, 'w') as f:
            # 转换datetime对象为字符串
            report['validation_time'] = str(report['validation_time'])
            json.dump(report, f, indent=2)

        print(f"\n验证报告已保存到: {report_path}")
    else:
        print(f"文件不存在: {csv_path}")


if __name__ == "__main__":
    main()

4.3.2 数据清洗策略

对于发现的问题，我们需要相应的清洗策略：

"""
数据清洗工具
"""
import pandas as pd
import numpy as np
from typing import Optional


def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    清洗数据的通用函数

    Parameters
    ----------
    df : pd.DataFrame
        原始数据

    Returns
    -------
    pd.DataFrame
        清洗后的数据
    """
    print("开始数据清洗...")
    original_rows = len(df)

    # 1. 删除完全重复的行
    df = df.drop_duplicates()

    # 2. 处理时间戳重复（保留第一条）
    if 'timestamp' in df.columns:
        df = df.drop_duplicates(subset=['timestamp'], keep='first')

    # 3. 处理空值
    numeric_columns = df.select_dtypes(include=[np.number]).columns
    df[numeric_columns] = df[numeric_columns].fillna(method='ffill').fillna(method='bfill')

    # 4. 修正价格关系
    if all(col in df.columns for col in ['open', 'high', 'low', 'close']):
        # 确保high是最大的
        df['high'] = df[['high', 'open', 'close']].max(axis=1)
        # 确保low是最小的
        df['low'] = df[['low', 'open', 'close']].min(axis=1)

    # 5. 处理异常值
    for col in ['open', 'high', 'low', 'close']:
        if col in df.columns:
            # 使用IQR方法检测异常值
            Q1 = df[col].quantile(0.25)
            Q3 = df[col].quantile(0.75)
            IQR = Q3 - Q1

            # 定义异常值范围
            lower_bound = Q1 - 1.5 * IQR
            upper_bound = Q3 + 1.5 * IQR

            # 限制异常值
            df[col] = df[col].clip(lower=lower_bound, upper_bound=upper_bound)

    # 6. 确保价格为正
    price_columns = ['open', 'high', 'low', 'close']
    for col in price_columns:
        if col in df.columns:
            df[col] = df[col].abs()

    cleaned_rows = len(df)
    print(f"数据清洗完成: {original_rows} -> {cleaned_rows} 行")

    return df

4.4 时间同步处理

在真实交易中，数据来自多个源，时间同步至关重要。

4.4.1 时区处理

"""
时间同步工具
处理不同时区和时间戳格式
"""

import pandas as pd
from datetime import datetime, timezone
import pytz


class TimeSynchronizer:
    """时间同步器"""

    def __init__(self, target_timezone: str = 'UTC'):
        """
        初始化时间同步器

        Parameters
        ----------
        target_timezone : str
            目标时区
        """
        self.target_tz = pytz.timezone(target_timezone)

    def convert_timezone(
        self,
        df: pd.DataFrame,
        timestamp_col: str,
        source_timezone: str
    ) -> pd.DataFrame:
        """
        转换时区

        Parameters
        ----------
        df : pd.DataFrame
            数据
        timestamp_col : str
            时间戳列名
        source_timezone : str
            源时区

        Returns
        -------
        pd.DataFrame
            转换后的数据
        """
        print(f"转换时区: {source_timezone} -> {self.target_tz}")

        source_tz = pytz.timezone(source_timezone)

        # 本地化时间戳
        df[timestamp_col] = df[timestamp_col].dt.tz_localize(source_tz)

        # 转换到目标时区
        df[timestamp_col] = df[timestamp_col].dt.tz_convert(self.target_tz)

        return df

    def align_time_series(
        self,
        df_list: list[pd.DataFrame],
        timestamp_col: str = 'timestamp'
    ) -> list[pd.DataFrame]:
        """
        对齐多个时间序列

        Parameters
        ----------
        df_list : list[pd.DataFrame]
            数据框列表
        timestamp_col : str
            时间戳列名

        Returns
        -------
        list[pd.DataFrame]
            对齐后的数据框列表
        """
        print("对齐时间序列...")

        # 找到所有时间戳的交集
        all_timestamps = None

        for df in df_list:
            timestamps = set(df[timestamp_col])
            if all_timestamps is None:
                all_timestamps = timestamps
            else:
                all_timestamps &= timestamps

        print(f"共同时间戳数量: {len(all_timestamps)}")

        # 过滤每个数据框
        aligned_dfs = []
        for df in df_list:
            aligned_df = df[df[timestamp_col].isin(all_timestamps)]
            aligned_dfs.append(aligned_df.sort_values(timestamp_col))

        return aligned_dfs

4.5 下一步

在本章的第一部分，我们学习了：

NautilusTrader 支持的数据类型
从 CSV 导入数据的完整流程
数据质量控制方法
时间同步处理

在下一部分，我们将学习：

Parquet 数据目录的使用
高效的数据存储和检索
数据供应商集成
实时数据处理

4.6 总结

关键要点

NautilusTrader 支持多种市场数据类型
CSV 导入需要仔细的数据验证和清洗
时间同步对于多源数据至关重要
高质量的数据是成功回测的基础

实践建议

始终验证导入的数据
保存数据验证报告
使用适当的数据清洗策略
考虑使用 Parquet 格式提高性能

4.7 参考资料

RD-Agent 教程 - 第三章：进阶使用技巧

Henry Lin — Wed, 04 Feb 2026 09:23:11 +0000

RD-Agent 教程 - 第三章：进阶使用技巧

3.1 量化金融代理深度解析

量化金融场景架构

RD-Agent 的量化金融代理 RD-Agent(Q) 是业界首个数据驱动的多代理量化策略开发框架。它通过协调因子-模型共同优化，实现了全栈式量化研究开发自动化。

核心架构设计

graph TB
    A[市场数据] --> B[数据预处理]
    B --> C[因子挖掘代理]
    B --> D[模型演进代理]
    C --> E[因子评估]
    D --> F[模型评估]
    E --> G[联合优化代理]
    F --> G
    G --> H[策略回测]
    H --> I[性能分析]
    I --> J{满足要求?}
    J -->|否| K[知识反馈]
    K --> C
    K --> D
    J -->|是| L[策略部署]

    style C fill:#e1f5fe
    style D fill:#f3e5f5
    style G fill:#e8f5e8
    style H fill:#fff3e0

Qlib 框架集成

RD-Agent(Q) 深度集成了微软的 Qlib 量化投资库，提供：

数据管理：

🗃️ 标准化金融数据格式
📈 实时和历史数据接口
🔄 数据自动更新机制
🧹 数据清洗和预处理

回测引擎：

📊 高性能向量化回测
💰 交易成本和滑点模拟
📋 多种组合构建策略
📈 风险指标计算

模型库：

🤖 机器学习模型集成
📊 传统量化因子
🧠 深度学习模型
🔄 模型集成策略

因子挖掘和优化

`rdagent fin_factor` 详解

因子挖掘代理专注于发现和优化有效的量化因子。

启动命令：

# 基本因子挖掘
rdagent fin_factor

# 指定配置文件
rdagent fin_factor --config factor_config.yaml

# 指定迭代次数
export FACTOR_MAX_LOOP=15
rdagent fin_factor

工作流程：

假设生成阶段

   # 基于市场理论和历史经验生成因子假设
   factor_hypotheses = [
       "技术指标动量因子",
       "财务质量因子",
       "市场情绪因子",
       "宏观经济因子"
   ]

因子实现阶段

   # 自动生成因子计算代码
   class MomentumFactor:
       def calculate(self, data):
           # 自动生成的动量因子计算逻辑
           return (data['close'] / data['close'].shift(20) - 1)

因子评估阶段

   # 多维度因子评估
   evaluation_metrics = {
       'IC': 0.045,          # 信息系数
       'IC_IR': 1.2,         # 信息比率
       'rank_IC': 0.038,     # 排序IC
       'turnover': 0.8,      # 换手率
       'max_drawdown': 0.15  # 最大回撤
   }

因子生成策略

技术分析因子：

technical_factors = {
    "momentum": [
        "price_momentum_5d", "price_momentum_20d",
        "volume_momentum", "volatility_momentum"
    ],
    "mean_reversion": [
        "rsi_divergence", "bollinger_position",
        "price_deviation"
    ],
    "trend": [
        "ma_trend", "macd_signal", "trend_strength"
    ]
}

基本面因子：

fundamental_factors = {
    "profitability": [
        "roe_trend", "roa_improvement",
        "gross_margin_stability"
    ],
    "growth": [
        "revenue_growth_consistency", "eps_growth_quality",
        "cash_flow_growth"
    ],
    "valuation": [
        "pe_relative", "pb_sector_adjusted",
        "ev_ebitda_normalized"
    ]
}

另类因子：

alternative_factors = {
    "sentiment": [
        "news_sentiment_score", "social_media_buzz",
        "analyst_revision_momentum"
    ],
    "network": [
        "supply_chain_strength", "industry_correlation",
        "peer_performance_influence"
    ]
}

因子有效性评估

评估框架：

class FactorEvaluator:
    def __init__(self, benchmark_data, start_date, end_date):
        self.data = benchmark_data
        self.start_date = start_date
        self.end_date = end_date

    def evaluate_factor(self, factor_values):
        """综合评估因子有效性"""
        metrics = {}

        # 1. 信息系数分析
        metrics['IC'] = self.calculate_IC(factor_values)
        metrics['IC_std'] = self.calculate_IC_stability(factor_values)
        metrics['IC_IR'] = metrics['IC'] / metrics['IC_std']

        # 2. 单调性检验
        metrics['monotonicity'] = self.test_monotonicity(factor_values)

        # 3. 换手率分析
        metrics['turnover'] = self.calculate_turnover(factor_values)

        # 4. 回测表现
        backtest_result = self.backtest_factor(factor_values)
        metrics.update(backtest_result)

        return metrics

模型演进和优化

`rdagent fin_model` 详解

模型演进代理专注于预测模型的自动化开发和优化。

支持的模型类型：

模型类别	具体模型	适用场景	特点
线性模型	LinearRegression, Ridge, Lasso	简单快速预测	可解释性强
树模型	XGBoost, LightGBM, CatBoost	表格数据建模	特征重要性
神经网络	MLP, TabNet, DeepFM	复杂关系建模	非线性拟合
时序模型	LSTM, GRU, Transformer	序列数据预测	时间依赖
集成模型	Stacking, Voting, Blending	提升预测精度	模型融合

模型架构搜索

自动化架构搜索：

class ModelArchitectureSearch:
    def __init__(self):
        self.search_space = {
            "n_layers": [2, 3, 4, 5],
            "hidden_dims": [64, 128, 256, 512],
            "dropout_rate": [0.1, 0.2, 0.3, 0.4],
            "activation": ["relu", "tanh", "gelu"],
            "optimizer": ["adam", "adamw", "sgd"],
            "learning_rate": [0.001, 0.01, 0.1]
        }

    def search_best_architecture(self, train_data, valid_data):
        """搜索最优模型架构"""
        best_config = None
        best_score = float('-inf')

        for config in self.generate_configurations():
            model = self.build_model(config)
            score = self.evaluate_model(model, train_data, valid_data)

            if score > best_score:
                best_score = score
                best_config = config

        return best_config, best_score

超参数优化

多种优化算法：

from optuna import create_study
from sklearn.model_selection import cross_val_score

class HyperparameterOptimizer:
    def __init__(self, model_class, param_space):
        self.model_class = model_class
        self.param_space = param_space

    def optimize_with_optuna(self, X_train, y_train, n_trials=100):
        """使用 Optuna 进行超参数优化"""
        def objective(trial):
            params = {}
            for param_name, param_range in self.param_space.items():
                if isinstance(param_range, list):
                    params[param_name] = trial.suggest_categorical(
                        param_name, param_range
                    )
                elif isinstance(param_range, tuple):
                    if isinstance(param_range[0], int):
                        params[param_name] = trial.suggest_int(
                            param_name, param_range[0], param_range[1]
                        )
                    else:
                        params[param_name] = trial.suggest_float(
                            param_name, param_range[0], param_range[1]
                        )

            model = self.model_class(**params)
            scores = cross_val_score(model, X_train, y_train, cv=5)
            return scores.mean()

        study = create_study(direction='maximize')
        study.optimize(objective, n_trials=n_trials)

        return study.best_params, study.best_value

因子模型联合优化

`rdagent fin_quant` 详解

联合优化代理是 RD-Agent(Q) 的核心创新，实现因子和模型的协同演进。

联合优化策略：

交替优化策略

   class AlternatingOptimization:
       def __init__(self, factor_agent, model_agent):
           self.factor_agent = factor_agent
           self.model_agent = model_agent

       def optimize(self, max_iterations=10):
           for i in range(max_iterations):
               # 固定模型，优化因子
               new_factors = self.factor_agent.evolve_factors(
                   current_model=self.current_model
               )

               # 固定因子，优化模型
               new_model = self.model_agent.evolve_model(
                   current_factors=new_factors
               )

               # 评估联合性能
               performance = self.evaluate_joint_performance(
                   new_factors, new_model
               )

               if performance > self.best_performance:
                   self.update_best_solution(new_factors, new_model)

多目标优化策略

   class MultiObjectiveOptimization:
       def __init__(self):
           self.objectives = [
               "maximize_sharpe_ratio",
               "minimize_max_drawdown",
               "maximize_information_ratio",
               "minimize_turnover"
           ]

       def pareto_optimization(self, population):
           """帕累托前沿优化"""
           pareto_front = []

           for individual in population:
               is_dominated = False
               for other in population:
                   if self.dominates(other, individual):
                       is_dominated = True
                       break

               if not is_dominated:
                   pareto_front.append(individual)

           return pareto_front

多目标优化算法

NSGA-II 实现：

class NSGA2Optimizer:
    def __init__(self, population_size=100, generations=50):
        self.population_size = population_size
        self.generations = generations

    def optimize(self, factor_space, model_space):
        # 初始化种群
        population = self.initialize_population()

        for generation in range(self.generations):
            # 评估适应度
            fitness_values = self.evaluate_population(population)

            # 非支配排序
            fronts = self.fast_non_dominated_sort(population, fitness_values)

            # 选择下一代
            next_population = self.select_next_generation(
                fronts, population, fitness_values
            )

            # 交叉和变异
            offspring = self.crossover_and_mutation(next_population)
            population = offspring

        return self.get_pareto_front(population)

协作演进机制

知识共享机制：

class KnowledgeSharing:
    def __init__(self):
        self.factor_knowledge_base = {}
        self.model_knowledge_base = {}
        self.interaction_patterns = {}

    def share_factor_insights(self, factor_id, insights):
        """分享因子洞察"""
        self.factor_knowledge_base[factor_id] = {
            'effectiveness': insights['ic_score'],
            'stability': insights['ic_std'],
            'best_models': insights['compatible_models'],
            'market_regimes': insights['effective_periods']
        }

    def share_model_insights(self, model_id, insights):
        """分享模型洞察"""
        self.model_knowledge_base[model_id] = {
            'architecture': insights['model_config'],
            'performance': insights['validation_score'],
            'best_factors': insights['important_features'],
            'hyperparameters': insights['optimal_params']
        }

    def get_recommendations(self, current_factors, current_model):
        """获取优化建议"""
        recommendations = {
            'factor_suggestions': [],
            'model_suggestions': [],
            'joint_strategies': []
        }

        # 基于历史知识生成建议
        for factor in current_factors:
            if factor in self.factor_knowledge_base:
                knowledge = self.factor_knowledge_base[factor]
                recommendations['model_suggestions'].extend(
                    knowledge['best_models']
                )

        return recommendations

金融报告分析

`rdagent fin_factor_report` 使用

从金融报告中自动提取量化因子是 RD-Agent 的独特能力。

支持的报告类型：

📋 年度报告 (10-K, Annual Report)
📊 季度报告 (10-Q, Quarterly Report)
📰 新闻公告 (8-K, Press Release)
📈 分析师报告 (Research Report)
🏛️ 监管文件 (SEC Filings)

使用示例：

# 基本报告分析
rdagent fin_factor_report --report-folder ./financial_reports

# 指定报告类型
rdagent fin_factor_report \
    --report-folder ./reports \
    --report-type annual \
    --language zh

# 批量处理
rdagent fin_factor_report \
    --report-folder ./reports \
    --batch-size 50 \
    --parallel-workers 4

报告解析和特征提取

文本预处理流水线：

class ReportPreprocessor:
    def __init__(self):
        self.nlp = spacy.load("en_core_web_sm")
        self.financial_terms = self.load_financial_dictionary()

    def preprocess_report(self, report_text):
        """预处理财务报告"""
        # 1. 文档清理
        cleaned_text = self.clean_document(report_text)

        # 2. 段落分割
        sections = self.extract_sections(cleaned_text)

        # 3. 实体识别
        entities = self.extract_financial_entities(sections)

        # 4. 情感分析
        sentiment_scores = self.analyze_sentiment(sections)

        return {
            'sections': sections,
            'entities': entities,
            'sentiment': sentiment_scores,
            'metadata': self.extract_metadata(report_text)
        }

因子提取策略：

class FinancialFactorExtractor:
    def __init__(self):
        self.factor_extractors = {
            'growth_factors': GrowthFactorExtractor(),
            'quality_factors': QualityFactorExtractor(),
            'sentiment_factors': SentimentFactorExtractor(),
            'risk_factors': RiskFactorExtractor()
        }

    def extract_factors(self, processed_report):
        """从处理后的报告中提取因子"""
        extracted_factors = {}

        for factor_type, extractor in self.factor_extractors.items():
            factors = extractor.extract(processed_report)
            extracted_factors[factor_type] = factors

        return extracted_factors

class GrowthFactorExtractor:
    def extract(self, report_data):
        """提取成长性因子"""
        growth_factors = {}

        # 收入增长质量
        revenue_mentions = self.find_revenue_discussions(report_data)
        growth_factors['revenue_growth_quality'] = self.analyze_growth_quality(
            revenue_mentions
        )

        # 市场扩张策略
        expansion_mentions = self.find_expansion_discussions(report_data)
        growth_factors['expansion_strategy_score'] = self.score_expansion_strategy(
            expansion_mentions
        )

        # 研发投入承诺
        rd_mentions = self.find_rd_discussions(report_data)
        growth_factors['rd_commitment_score'] = self.analyze_rd_commitment(
            rd_mentions
        )

        return growth_factors

情感分析和信号挖掘

多维度情感分析：

class FinancialSentimentAnalyzer:
    def __init__(self):
        self.sentiment_model = pipeline(
            "sentiment-analysis",
            model="ProsusAI/finbert"
        )
        self.uncertainty_detector = UncertaintyDetector()
        self.forward_looking_detector = ForwardLookingDetector()

    def analyze_comprehensive_sentiment(self, text_sections):
        """综合情感分析"""
        sentiment_scores = {}

        for section_name, text in text_sections.items():
            # 基础情感分析
            basic_sentiment = self.sentiment_model(text)

            # 不确定性检测
            uncertainty_score = self.uncertainty_detector.detect(text)

            # 前瞻性语句检测
            forward_looking = self.forward_looking_detector.detect(text)

            sentiment_scores[section_name] = {
                'polarity': basic_sentiment[0]['score'],
                'label': basic_sentiment[0]['label'],
                'uncertainty': uncertainty_score,
                'forward_looking': forward_looking,
                'confidence': self.calculate_confidence(
                    basic_sentiment, uncertainty_score
                )
            }

        return sentiment_scores

class SignalGenerator:
    def __init__(self):
        self.signal_weights = {
            'management_tone': 0.3,
            'financial_health': 0.4,
            'market_position': 0.2,
            'future_outlook': 0.1
        }

    def generate_trading_signals(self, factor_scores, sentiment_scores):
        """生成交易信号"""
        composite_score = 0.0

        # 加权合成得分
        for factor_type, weight in self.signal_weights.items():
            if factor_type in factor_scores:
                composite_score += factor_scores[factor_type] * weight

        # 情感调整
        sentiment_adjustment = self.calculate_sentiment_adjustment(
            sentiment_scores
        )

        final_score = composite_score * sentiment_adjustment

        # 生成信号
        if final_score > 0.6:
            signal = "STRONG_BUY"
        elif final_score > 0.3:
            signal = "BUY"
        elif final_score > -0.3:
            signal = "HOLD"
        elif final_score > -0.6:
            signal = "SELL"
        else:
            signal = "STRONG_SELL"

        return {
            'signal': signal,
            'confidence': abs(final_score),
            'score': final_score,
            'components': {
                'factor_score': composite_score,
                'sentiment_adjustment': sentiment_adjustment
            }
        }

3.2 CoSTEER 演进框架深度解析

演进框架核心原理

CoSTEER (Collaborative evolving STrategy for automatic data-cEntric dEvelopment fRamework) 是 RD-Agent 的核心演进引擎，实现了智能体的自主学习和持续改进。

协作演进策略设计

CoSTEER 框架基于以下核心原理：

多代理协作 - 不同专业代理协同工作
知识积累 - 从历史经验中学习和改进
自适应演进 - 根据反馈动态调整策略
RAG增强 - 检索增强生成提升决策质量

class CoSTEER(Developer[Experiment]):
    """CoSTEER 演进框架核心类"""

    def __init__(
        self,
        settings: CoSTEERSettings,
        eva: RAGEvaluator,
        es: EvolvingStrategy,
        evolving_version: int = 2,
        with_knowledge: bool = True,
        knowledge_self_gen: bool = True,
        max_loop: int | None = None,
    ):
        self.settings = settings
        self.max_loop = settings.max_loop if max_loop is None else max_loop
        self.knowledge_base_path = Path(settings.knowledge_base_path)
        self.with_knowledge = with_knowledge
        self.knowledge_self_gen = knowledge_self_gen
        self.evolving_strategy = es
        self.evaluator = eva

        # 初始化 RAG 系统
        self.rag_strategy = self._init_rag_strategy(evolving_version)

知识管理机制

多层次知识表示：

class KnowledgeManagementSystem:
    def __init__(self):
        self.knowledge_layers = {
            'factual_knowledge': FactualKnowledgeBase(),
            'procedural_knowledge': ProceduralKnowledgeBase(),
            'experiential_knowledge': ExperientialKnowledgeBase(),
            'meta_knowledge': MetaKnowledgeBase()
        }

    def store_knowledge(self, knowledge_item):
        """分层存储知识"""
        knowledge_type = self.classify_knowledge(knowledge_item)
        self.knowledge_layers[knowledge_type].store(knowledge_item)

    def retrieve_relevant_knowledge(self, query_context):
        """检索相关知识"""
        relevant_knowledge = {}

        for layer_name, knowledge_base in self.knowledge_layers.items():
            relevant_items = knowledge_base.query(
                query_context,
                similarity_threshold=0.7
            )
            relevant_knowledge[layer_name] = relevant_items

        return self.synthesize_knowledge(relevant_knowledge)

知识表示格式：

@dataclass
class KnowledgeItem:
    """知识项数据结构"""
    id: str
    content: str
    knowledge_type: KnowledgeType
    domain: str
    confidence: float
    source: str
    timestamp: datetime
    usage_count: int = 0
    success_rate: float = 0.0

    def to_embedding(self) -> np.ndarray:
        """转换为向量表示"""
        return self.embedding_model.encode(self.content)

    def update_effectiveness(self, success: bool):
        """更新知识有效性"""
        self.usage_count += 1
        if success:
            self.success_rate = (
                (self.success_rate * (self.usage_count - 1) + 1.0)
                / self.usage_count
            )
        else:
            self.success_rate = (
                self.success_rate * (self.usage_count - 1)
                / self.usage_count
            )

自适应优化算法

多策略演进算法：

class AdaptiveEvolutionStrategy:
    def __init__(self):
        self.evolution_strategies = {
            'genetic_algorithm': GeneticAlgorithmStrategy(),
            'differential_evolution': DifferentialEvolutionStrategy(),
            'particle_swarm': ParticleSwarmStrategy(),
            'simulated_annealing': SimulatedAnnealingStrategy()
        }
        self.strategy_performance = {}
        self.adaptive_weights = {}

    def evolve_population(self, population, generation):
        """自适应演进种群"""
        # 选择演进策略
        selected_strategies = self.select_strategies(generation)

        new_population = []
        for strategy_name, weight in selected_strategies.items():
            strategy = self.evolution_strategies[strategy_name]

            # 按权重分配种群
            sub_population_size = int(len(population) * weight)
            sub_population = population[:sub_population_size]

            # 演进子种群
            evolved_sub_pop = strategy.evolve(sub_population)
            new_population.extend(evolved_sub_pop)

        # 更新策略性能
        self.update_strategy_performance(new_population, generation)

        return new_population

    def select_strategies(self, generation):
        """基于历史性能选择演进策略"""
        if generation < 5:
            # 初期均匀分配
            return {name: 1.0/len(self.evolution_strategies)
                   for name in self.evolution_strategies.keys()}

        # 基于性能调整权重
        total_performance = sum(self.strategy_performance.values())
        return {
            name: performance / total_performance
            for name, performance in self.strategy_performance.items()
        }

自定义演进策略

策略配置和调整

配置文件示例：

# costeer_config.yaml
costeer_settings:
  max_loop: 10
  knowledge_base_path: "./knowledge_base"
  new_knowledge_base_path: "./new_knowledge"
  evolving_version: 2

evolution_strategy:
  population_size: 50
  mutation_rate: 0.1
  crossover_rate: 0.8
  selection_method: "tournament"
  tournament_size: 5

rag_settings:
  embedding_model: "text-embedding-3-small"
  similarity_threshold: 0.75
  max_retrieved_docs: 10
  rerank_top_k: 5

evaluation_settings:
  metrics: ["accuracy", "f1_score", "auc_roc"]
  cross_validation_folds: 5
  test_ratio: 0.2

knowledge_management:
  auto_cleanup: true
  cleanup_threshold: 0.3
  knowledge_update_frequency: 100
  max_knowledge_items: 10000

自定义演进策略实现：

class CustomEvolutionStrategy(EvolvingStrategy):
    """自定义演进策略"""

    def __init__(self, config: dict):
        super().__init__()
        self.config = config
        self.population_size = config.get('population_size', 50)
        self.mutation_rate = config.get('mutation_rate', 0.1)
        self.crossover_rate = config.get('crossover_rate', 0.8)

    def initialize_population(self, size: int = None):
        """初始化种群"""
        size = size or self.population_size
        population = []

        for _ in range(size):
            individual = self.create_random_individual()
            population.append(individual)

        return population

    def evolve_generation(self, population, fitness_scores):
        """演进一代"""
        new_population = []

        # 精英选择 - 保留最优个体
        elite_count = int(len(population) * 0.1)
        elite_indices = np.argsort(fitness_scores)[-elite_count:]
        for idx in elite_indices:
            new_population.append(population[idx])

        # 交叉和变异生成新个体
        while len(new_population) < len(population):
            # 选择父代
            parent1 = self.tournament_selection(population, fitness_scores)
            parent2 = self.tournament_selection(population, fitness_scores)

            # 交叉
            if random.random() < self.crossover_rate:
                child1, child2 = self.crossover(parent1, parent2)
            else:
                child1, child2 = parent1, parent2

            # 变异
            if random.random() < self.mutation_rate:
                child1 = self.mutate(child1)
            if random.random() < self.mutation_rate:
                child2 = self.mutate(child2)

            new_population.extend([child1, child2])

        return new_population[:len(population)]

    def crossover(self, parent1, parent2):
        """交叉操作"""
        # 实现具体的交叉逻辑
        # 这里以参数交叉为例
        child1_params = {}
        child2_params = {}

        for key in parent1.parameters.keys():
            if random.random() < 0.5:
                child1_params[key] = parent1.parameters[key]
                child2_params[key] = parent2.parameters[key]
            else:
                child1_params[key] = parent2.parameters[key]
                child2_params[key] = parent1.parameters[key]

        child1 = parent1.copy()
        child2 = parent2.copy()
        child1.parameters = child1_params
        child2.parameters = child2_params

        return child1, child2

    def mutate(self, individual):
        """变异操作"""
        mutated = individual.copy()

        for key, value in mutated.parameters.items():
            if random.random() < self.mutation_rate:
                if isinstance(value, (int, float)):
                    # 高斯变异
                    noise = np.random.normal(0, 0.1 * abs(value))
                    mutated.parameters[key] = value + noise
                elif isinstance(value, str):
                    # 字符串变异（如模型名称）
                    mutated.parameters[key] = self.mutate_string(value)

        return mutated

评估器自定义

多指标评估器：

class MultiMetricEvaluator(RAGEvaluator):
    """多指标评估器"""

    def __init__(self, metrics_config: dict):
        super().__init__()
        self.metrics_config = metrics_config
        self.metric_weights = metrics_config.get('weights', {})
        self.metric_calculators = self._init_metric_calculators()

    def evaluate(self, experiment: Experiment) -> Dict[str, float]:
        """多指标评估"""
        metrics = {}

        for metric_name, calculator in self.metric_calculators.items():
            try:
                score = calculator.calculate(experiment)
                metrics[metric_name] = score
            except Exception as e:
                logger.warning(f"计算指标 {metric_name} 失败: {e}")
                metrics[metric_name] = 0.0

        # 计算加权综合得分
        weighted_score = self._calculate_weighted_score(metrics)
        metrics['weighted_score'] = weighted_score

        return metrics

    def _calculate_weighted_score(self, metrics: Dict[str, float]) -> float:
        """计算加权综合得分"""
        total_score = 0.0
        total_weight = 0.0

        for metric_name, score in metrics.items():
            if metric_name in self.metric_weights:
                weight = self.metric_weights[metric_name]
                total_score += score * weight
                total_weight += weight

        return total_score / total_weight if total_weight > 0 else 0.0

class CustomMetricCalculator:
    """自定义指标计算器"""

    def __init__(self, metric_name: str, calculation_func: callable):
        self.metric_name = metric_name
        self.calculation_func = calculation_func

    def calculate(self, experiment: Experiment) -> float:
        """计算指标值"""
        return self.calculation_func(experiment)

# 使用示例
def sharpe_ratio_calculator(experiment):
    """计算夏普比率"""
    returns = experiment.get_returns()
    if len(returns) == 0 or returns.std() == 0:
        return 0.0
    return returns.mean() / returns.std() * np.sqrt(252)

def max_drawdown_calculator(experiment):
    """计算最大回撤"""
    cumulative_returns = (1 + experiment.get_returns()).cumprod()
    running_max = cumulative_returns.expanding().max()
    drawdown = (cumulative_returns - running_max) / running_max
    return -drawdown.min()  # 返回正值

# 创建自定义评估器
metrics_config = {
    'weights': {
        'sharpe_ratio': 0.4,
        'max_drawdown': 0.3,
        'information_ratio': 0.3
    }
}

evaluator = MultiMetricEvaluator(metrics_config)
evaluator.add_metric_calculator('sharpe_ratio', sharpe_ratio_calculator)
evaluator.add_metric_calculator('max_drawdown', max_drawdown_calculator)

知识库管理

知识库优化策略：

class KnowledgeBaseManager:
    """知识库管理器"""

    def __init__(self, config: dict):
        self.config = config
        self.knowledge_store = {}
        self.embedding_cache = {}
        self.access_frequency = defaultdict(int)
        self.last_access_time = {}

    def add_knowledge(self, knowledge_item: KnowledgeItem):
        """添加知识项"""
        # 检查知识库大小限制
        if len(self.knowledge_store) >= self.config.get('max_items', 10000):
            self._cleanup_knowledge()

        self.knowledge_store[knowledge_item.id] = knowledge_item
        self.embedding_cache[knowledge_item.id] = knowledge_item.to_embedding()
        self.last_access_time[knowledge_item.id] = datetime.now()

    def query_knowledge(
        self,
        query: str,
        top_k: int = 10,
        similarity_threshold: float = 0.7
    ) -> List[KnowledgeItem]:
        """查询相关知识"""
        query_embedding = self._get_query_embedding(query)
        similarities = {}

        for item_id, item_embedding in self.embedding_cache.items():
            similarity = cosine_similarity(
                query_embedding.reshape(1, -1),
                item_embedding.reshape(1, -1)
            )[0][0]

            if similarity >= similarity_threshold:
                similarities[item_id] = similarity
                self.access_frequency[item_id] += 1
                self.last_access_time[item_id] = datetime.now()

        # 按相似度排序并返回top-k
        sorted_items = sorted(
            similarities.items(),
            key=lambda x: x[1],
            reverse=True
        )[:top_k]

        return [self.knowledge_store[item_id] for item_id, _ in sorted_items]

    def _cleanup_knowledge(self):
        """清理知识库"""
        cleanup_count = int(len(self.knowledge_store) * 0.1)  # 清理10%

        # 基于访问频率和时间的综合评分
        scores = {}
        current_time = datetime.now()

        for item_id, knowledge_item in self.knowledge_store.items():
            frequency_score = self.access_frequency[item_id]
            time_score = (
                current_time - self.last_access_time[item_id]
            ).days
            effectiveness_score = knowledge_item.success_rate

            # 综合评分（频率高、时间近、效果好的得分高）
            composite_score = (
                frequency_score * 0.4 +
                effectiveness_score * 0.4 -
                time_score * 0.2
            )
            scores[item_id] = composite_score

        # 删除评分最低的项目
        items_to_remove = sorted(
            scores.items(),
            key=lambda x: x[1]
        )[:cleanup_count]

        for item_id, _ in items_to_remove:
            del self.knowledge_store[item_id]
            del self.embedding_cache[item_id]
            del self.access_frequency[item_id]
            del self.last_access_time[item_id]

        logger.info(f"清理了 {cleanup_count} 个知识项")

多代理协作机制

研究代理职责

class ResearchAgent:
    """研究代理 - 负责假设生成和策略规划"""

    def __init__(self, knowledge_base: KnowledgeBase):
        self.knowledge_base = knowledge_base
        self.hypothesis_generator = HypothesisGenerator()
        self.strategy_planner = StrategyPlanner()

    def generate_hypotheses(self, problem_context: dict) -> List[Hypothesis]:
        """生成研究假设"""
        # 从知识库检索相关理论
        relevant_knowledge = self.knowledge_base.query(
            problem_context['description']
        )

        # 基于领域知识生成假设
        hypotheses = []
        for knowledge_item in relevant_knowledge:
            hypothesis = self.hypothesis_generator.generate_from_knowledge(
                knowledge_item, problem_context
            )
            hypotheses.append(hypothesis)

        # 生成创新假设
        novel_hypotheses = self.hypothesis_generator.generate_novel(
            problem_context, existing_hypotheses=hypotheses
        )
        hypotheses.extend(novel_hypotheses)

        return self.rank_hypotheses(hypotheses)

    def plan_research_strategy(self, hypothesis: Hypothesis) -> ResearchPlan:
        """规划研究策略"""
        return self.strategy_planner.create_plan(
            hypothesis=hypothesis,
            available_resources=self.get_available_resources(),
            time_constraints=self.get_time_constraints()
        )

开发代理功能

class DevelopmentAgent:
    """开发代理 - 负责代码实现和测试验证"""

    def __init__(self, code_generator: CodeGenerator, validator: CodeValidator):
        self.code_generator = code_generator
        self.validator = validator
        self.implementation_cache = {}

    def implement_hypothesis(self, hypothesis: Hypothesis) -> Implementation:
        """实现研究假设"""
        # 生成实现代码
        implementation_code = self.code_generator.generate_code(
            specification=hypothesis.specification,
            constraints=hypothesis.constraints
        )

        # 验证代码正确性
        validation_result = self.validator.validate(implementation_code)

        if not validation_result.is_valid:
            # 修复代码问题
            fixed_code = self.code_generator.fix_issues(
                implementation_code,
                validation_result.issues
            )
            implementation_code = fixed_code

        # 创建实现对象
        implementation = Implementation(
            code=implementation_code,
            hypothesis=hypothesis,
            validation_result=validation_result
        )

        # 缓存实现
        self.implementation_cache[hypothesis.id] = implementation

        return implementation

    def optimize_implementation(
        self,
        implementation: Implementation,
        feedback: Feedback
    ) -> Implementation:
        """基于反馈优化实现"""
        optimization_suggestions = self.analyze_feedback(feedback)

        optimized_code = self.code_generator.optimize_code(
            implementation.code,
            optimization_suggestions
        )

        return Implementation(
            code=optimized_code,
            hypothesis=implementation.hypothesis,
            parent_implementation=implementation
        )

代理间通信协议

class AgentCommunicationProtocol:
    """代理间通信协议"""

    def __init__(self):
        self.message_queue = {}
        self.subscriptions = defaultdict(list)
        self.message_handlers = {}

    def register_agent(self, agent_id: str, agent: Agent):
        """注册代理"""
        self.message_queue[agent_id] = []
        agent.set_communication_protocol(self)

    def subscribe(self, agent_id: str, message_type: str):
        """订阅消息类型"""
        self.subscriptions[message_type].append(agent_id)

    def publish(self, sender_id: str, message_type: str, message: dict):
        """发布消息"""
        # 添加消息元数据
        full_message = {
            'sender': sender_id,
            'type': message_type,
            'timestamp': datetime.now(),
            'content': message
        }

        # 发送给所有订阅者
        for subscriber_id in self.subscriptions[message_type]:
            if subscriber_id != sender_id:  # 不发送给自己
                self.message_queue[subscriber_id].append(full_message)

    def get_messages(self, agent_id: str) -> List[dict]:
        """获取代理的消息"""
        messages = self.message_queue.get(agent_id, [])
        self.message_queue[agent_id] = []  # 清空队列
        return messages

class CollaborativeWorkflow:
    """协作工作流"""

    def __init__(self, agents: List[Agent], protocol: AgentCommunicationProtocol):
        self.agents = {agent.id: agent for agent in agents}
        self.protocol = protocol
        self.workflow_state = WorkflowState.INITIALIZED

    def execute_collaborative_task(self, task: CollaborativeTask):
        """执行协作任务"""
        # 分解任务
        subtasks = self.decompose_task(task)

        # 分配子任务给合适的代理
        task_assignments = self.assign_tasks(subtasks)

        # 执行子任务
        results = {}
        for agent_id, assigned_tasks in task_assignments.items():
            agent = self.agents[agent_id]
            agent_results = agent.execute_tasks(assigned_tasks)
            results[agent_id] = agent_results

        # 协调和整合结果
        integrated_result = self.integrate_results(results)

        return integrated_result

    def coordinate_agents(self):
        """协调代理活动"""
        # 处理代理间消息
        for agent_id, agent in self.agents.items():
            messages = self.protocol.get_messages(agent_id)
            if messages:
                agent.process_messages(messages)

        # 检查是否需要协调
        coordination_needed = self.check_coordination_needs()
        if coordination_needed:
            self.perform_coordination()

本章内容展示了 RD-Agent 的高级功能和深度定制能力，包括量化金融代理的专业功能、CoSTEER 演进框架的核心机制，以及多代理协作的实现方法。下一章我们将通过实战案例来演示这些高级功能的具体应用。

RD-Agent 教程 - 第二章：核心功能详解

Henry Lin — Wed, 04 Feb 2026 09:22:17 +0000

RD-Agent 教程 - 第二章：核心功能详解

2.1 命令行界面基础

CLI 结构和设计原理

RD-Agent 采用统一的命令行界面（CLI）设计，所有功能都通过 rdagent 命令访问。CLI 的核心实现位于 rdagent/app/cli.py，使用 Typer 框架构建，提供了直观和强大的命令行体验。

架构分析

# rdagent/app/cli.py 核心结构
import typer
from rdagent.app.data_science.loop import main as data_science
from rdagent.app.qlib_rd_loop.factor import main as fin_factor
from rdagent.app.qlib_rd_loop.model import main as fin_model
from rdagent.app.qlib_rd_loop.quant import main as fin_quant

app = typer.Typer()

# 注册命令
app.command(name="fin_factor")(fin_factor)
app.command(name="fin_model")(fin_model)
app.command(name="data_science")(data_science)

命令分类和组织方式

RD-Agent 的命令按功能领域组织：

命令类别	命令	功能描述	主要用途
金融量化	`fin_factor`	因子自动挖掘	发现有效的量化因子
	`fin_model`	模型自动演进	优化预测模型
	`fin_quant`	因子模型联合优化	端到端策略开发
	`fin_factor_report`	报告因子提取	从金融报告提取信号
数据科学	`data_science`	通用数据科学任务	ML 竞赛、建模项目
通用模型	`general_model`	论文模型实现	研究论文复现
工具功能	`ui`	Web 界面	可视化监控
	`health_check`	健康检查	系统状态验证
	`collect_info`	信息收集	诊断和调试

自动环境加载机制

RD-Agent CLI 的一个重要特性是自动加载环境变量：

from dotenv import load_dotenv
load_dotenv(".env")  # 自动加载当前目录的 .env 文件

优势：

🔄 自动配置加载，无需手动设置环境变量
📁 支持项目级配置，不同项目可有不同配置
🔒 配置文件本地存储，提高安全性

基础命令使用

帮助系统使用

# 查看所有可用命令
rdagent --help

# 查看特定命令的详细帮助
rdagent fin_factor --help
rdagent data_science --help

# 查看子命令参数
rdagent ui --help

示例输出：

Usage: rdagent [OPTIONS] COMMAND [ARGS]...

Options:
  --help  Show this message and exit.

Commands:
  data_science        Run data science automation
  fin_factor         Run factor mining
  fin_model          Run model evolution
  fin_quant          Run factor-model joint optimization
  general_model      Extract and implement models from papers
  health_check       Check system health
  ui                 Start web interface

全局参数和选项

虽然 RD-Agent 的大部分配置通过 .env 文件管理，但某些命令支持运行时参数：

# UI 命令的常用参数
rdagent ui --port 19899 --log-dir ./log --debug

# 健康检查的选择性检查
rdagent health_check --no-check-env --no-check-docker

# 数据科学命令的竞赛指定
rdagent data_science --competition tabular-playground-series-dec-2021

日志和输出管理

RD-Agent 提供丰富的日志记录功能：

# 默认日志级别运行
rdagent fin_factor

# 启用详细日志（通过环境变量）
export RDAGENT_LOG_LEVEL=DEBUG
rdagent fin_factor

# 指定日志输出目录
export RDAGENT_LOG_DIR=./custom_logs
rdagent fin_quant

# 快速测试，只验证一次
rdagent data_science --competition titanic --loop-n 1

日志结构：

logs/
├── rdagent.log                 # 主日志文件
├── experiments/                # 实验日志
│   ├── factor_exp_001/
│   └── model_exp_001/
└── ui/                        # Web UI 日志
    └── access.log

监控界面使用

Web UI 启动和配置

RD-Agent 提供了强大的 Web 界面用于实时监控和结果查看：

# 基本启动
rdagent ui --port 19899

# 指定日志目录
rdagent ui --port 19899 --log-dir ./logs

# 启用调试模式
rdagent ui --port 19899 --debug

# 数据科学专用界面
rdagent ui --port 19899 --data_science True

实时监控功能

Web UI 提供以下监控能力：

1. 实验进度追踪

📊 实时显示实验状态
📈 性能指标图表
⏱️ 时间线视图
🔄 自动刷新

2. 日志查看器

📝 结构化日志显示
🔍 日志搜索和过滤
📋 多级别日志（INFO、DEBUG、ERROR）
💾 日志导出功能

3. 结果可视化

📊 因子性能对比
📈 模型训练曲线
🎯 回测结果图表
📋 详细性能报告

界面功能详解

主仪表板：

┌─────────────────────────────────────────┐
│ RD-Agent 监控面板                        │
├─────────────────────────────────────────┤
│ 🔄 实验状态: 运行中 (2/5)                │
│ ⏱️ 运行时间: 1小时23分钟                  │
│ 📊 当前任务: 因子验证                     │
│ 🎯 最佳得分: 0.234 (IC)                  │
└─────────────────────────────────────────┘

实验列表：

实验ID    | 类型    | 状态    | 开始时间    | 得分
---------|---------|---------|-------------|-------
EXP_001  | 因子    | 完成    | 14:23:01   | 0.198
EXP_002  | 因子    | 运行中  | 14:45:12   | -
EXP_003  | 模型    | 等待    | -          | -

2.2 数据科学代理详解

数据科学场景概述

RD-Agent 的数据科学代理是一个全自动的机器学习工程系统，能够自主完成从数据探索到模型部署的完整流程。该代理在 MLE-bench 基准测试中获得第一名，展现了强大的自动化能力。

支持的任务类型

任务类型	描述	应用场景	技术特点
表格数据建模	结构化数据的预测任务	金融预测、用户行为分析	特征工程、模型集成
时间序列预测	时序数据的预测建模	股价预测、需求预测	时序特征、序列模型
图像分类	计算机视觉任务	医疗影像、产品识别	CNN、迁移学习
自然语言处理	文本数据处理	情感分析、文档分类	预训练模型、嵌入
回归分析	连续值预测	价格预测、评分预估	线性/非线性模型

工作流程和架构

graph TD
    A[数据输入] --> B[假设生成]
    B --> C[实验设计]
    C --> D[特征工程]
    D --> E[模型开发]
    E --> F[验证测试]
    F --> G[结果分析]
    G --> H{性能满足要求?}
    H -->|否| I[反馈学习]
    I --> B
    H -->|是| J[模型输出]

    style B fill:#e1f5fe
    style D fill:#f3e5f5
    style E fill:#e8f5e8
    style G fill:#fff3e0

核心组件：

假设生成器 - 基于数据特征和问题类型生成建模假设
实验管理器 - 系统化管理实验流程和资源
特征工程器 - 自动生成和选择有效特征
模型开发器 - 自动选择和调优模型
评估系统 - 多维度评估模型性能
知识管理 - 积累和复用实验经验

与其他工具的集成

# 支持的数据格式和来源
Data Sources:
├── CSV/Excel 文件
├── Kaggle 竞赛数据
├── 数据库连接 (SQL)
├── API 数据源
├── 图像文件夹
└── 文本文档集合

Model Frameworks:
├── Scikit-learn
├── XGBoost/LightGBM
├── PyTorch
├── TensorFlow/Keras
├── Transformers (Hugging Face)
└── 自定义模型

Kaggle 竞赛自动化

Kaggle API 配置

首先需要配置 Kaggle API 以访问竞赛数据：

# 1. 获取 Kaggle API Token
# 登录 Kaggle -> Account -> Create New Token
# 下载 kaggle.json 文件

# 2. 配置 API 文件
mkdir -p ~/.kaggle
mv ~/Downloads/kaggle.json ~/.kaggle/
chmod 600 ~/.kaggle/kaggle.json

# 3. 验证配置
kaggle competitions list

环境变量配置

在 .env 文件中添加数据科学相关配置：

# 数据科学代理配置
DS_LOCAL_DATA_PATH="./data/competitions"
DS_CODER_ON_WHOLE_PIPELINE=True
DS_IF_USING_MLE_DATA=True
DS_SAMPLE_DATA_BY_LLM=True
DS_SCEN=rdagent.scenarios.data_science.scen.KaggleScen

# LLM 配置（如前所述）
CHAT_MODEL=gpt-4o
EMBEDDING_MODEL=text-embedding-3-small
OPENAI_API_KEY=your-api-key

竞赛数据自动下载

# 自动下载并处理竞赛数据
rdagent data_science --competition tabular-playground-series-dec-2021

# 指定自定义数据路径
export DS_LOCAL_DATA_PATH="./custom_data"
rdagent data_science --competition house-prices-advanced-regression-techniques

完整竞赛流程演示

以 "Tabular Playground Series - Dec 2021" 为例：

Step 1: 环境准备

# 创建项目目录
mkdir kaggle_rdagent_demo && cd kaggle_rdagent_demo

# 配置环境
cat > .env << EOF
CHAT_MODEL=gpt-4o
EMBEDDING_MODEL=text-embedding-3-small
OPENAI_API_KEY=your-openai-api-key

DS_LOCAL_DATA_PATH="$(pwd)/data"
DS_CODER_ON_WHOLE_PIPELINE=True
DS_IF_USING_MLE_DATA=True
DS_SAMPLE_DATA_BY_LLM=True
DS_SCEN=rdagent.scenarios.data_science.scen.KaggleScen
EOF

Step 2: 启动自动化流程

# 启动 RD-Agent 数据科学代理
rdagent data_science --competition tabular-playground-series-dec-2021

Step 3: 实时监控

# 在另一个终端启动监控界面
rdagent ui --port 19899 --data_science True
# 访问 http://localhost:19899

自动执行的流程：

数据下载和探索
- 自动下载竞赛数据
- 生成数据探索报告
- 识别数据类型和分布
假设生成
- 基于数据特征生成建模假设
- 分析目标变量特性
- 识别潜在的特征工程机会
特征工程
- 自动生成数值特征变换
- 类别特征编码
- 交互特征创建
- 时间特征提取（如适用）
模型开发
- 尝试多种算法（随机森林、XGBoost、神经网络等）
- 自动超参数调优
- 交叉验证和模型选择
模型集成
- Stacking/Blending 策略
- 多模型融合
- 权重优化
提交准备
- 生成预测文件
- 格式验证
- 自动提交（可选）

医疗预测模型场景

医疗数据处理

医疗数据通常具有特殊性，RD-Agent 提供专门的处理能力：

# 运行医疗预测任务
rdagent data_science --competition medical-prediction-task

# 示例：急性肾衰竭预测
wget https://github.com/SunsetWolf/rdagent_resource/releases/download/ds_data/arf-12-hours-prediction-task.zip
unzip arf-12-hours-prediction-task.zip -d ./data/

医疗数据特点处理：

隐私保护
- 数据脱敏和匿名化
- HIPAA 合规性考虑
- 本地处理，不上传原始数据
不平衡数据处理
- SMOTE 过采样
- 成本敏感学习
- 阈值调优
时序医疗数据
- 病程时间建模
- 多时间点特征
- 生存分析方法
多模态数据融合
- 结构化数据（检验指标）
- 非结构化数据（病历文本）
- 医学图像数据

预测模型开发

# 医疗预测的特殊配置示例
medical_config = {
    "task_type": "classification",
    "positive_class_weight": 10,  # 处理不平衡
    "cross_validation_folds": 10,  # 更多折数
    "feature_selection": "medical_relevance",
    "interpretability": True,  # 医疗需要可解释性
    "ensemble_methods": ["voting", "stacking"],
    "evaluation_metrics": ["auc", "precision", "recall", "f1"],
}

模型评估和优化

医疗模型的评估更注重：

🎯 临床指标: Sensitivity, Specificity, PPV, NPV
📊 ROC/PR 曲线: 不同阈值下的性能
⚖️ 成本效益分析: 误诊和漏诊的代价
🔍 可解释性: SHAP 值和特征重要性

通用数据科学任务

自定义数据集处理

对于非竞赛数据，RD-Agent 支持灵活的数据接入：

# 创建数据目录结构
mkdir -p custom_project/data
cd custom_project

# 准备数据文件
# train.csv - 训练数据
# test.csv - 测试数据（可选）
# sample_submission.csv - 提交格式（可选）

# 配置环境
cat > .env << EOF
DS_LOCAL_DATA_PATH="$(pwd)/data"
DS_CODER_ON_WHOLE_PIPELINE=True
DS_IF_USING_MLE_DATA=False  # 非 MLE 数据
DS_SAMPLE_DATA_BY_LLM=False
DS_SCEN=rdagent.scenarios.data_science.scen.DataScienceScen
EOF

# 运行分析
rdagent data_science --competition custom_project

数据格式要求

标准表格数据：

data/
├── train.csv          # 必需：训练数据
├── test.csv           # 可选：测试数据
├── sample_submission.csv  # 可选：提交格式
└── description.md     # 可选：问题描述

多模态数据：

data/
├── tabular/
│   ├── train.csv
│   └── test.csv
├── images/
│   ├── train/
│   └── test/
├── text/
│   ├── train_texts.json
│   └── test_texts.json
└── config.yaml       # 数据配置文件

模型选择和调优

RD-Agent 智能选择最适合的算法：

分类任务算法选择：

classification_algorithms = {
    "small_data": ["RandomForest", "SVM", "LogisticRegression"],
    "medium_data": ["XGBoost", "LightGBM", "CatBoost"],
    "large_data": ["NeuralNetwork", "TabNet", "AutoML"],
    "text_data": ["BERT", "RoBERTa", "DistilBERT"],
    "image_data": ["ResNet", "EfficientNet", "ViT"],
}

超参数调优策略：

tuning_strategies = {
    "bayesian_optimization": "高效的参数空间搜索",
    "random_search": "快速初步调优",
    "grid_search": "精确但耗时的搜索",
    "evolutionary": "复杂空间的全局优化",
    "hyperband": "多保真度优化",
}

特征工程自动化

数值特征处理：

numerical_transformations = [
    "StandardScaler",      # 标准化
    "MinMaxScaler",        # 归一化
    "RobustScaler",        # 鲁棒缩放
    "PowerTransformer",    # 幂变换
    "QuantileTransformer", # 分位数变换
    "PCA",                 # 主成分分析
    "PolynomialFeatures",  # 多项式特征
]

类别特征处理：

categorical_transformations = [
    "OneHotEncoder",       # 独热编码
    "LabelEncoder",        # 标签编码
    "TargetEncoder",       # 目标编码
    "BinaryEncoder",       # 二进制编码
    "HashingEncoder",      # 哈希编码
    "FrequencyEncoder",    # 频次编码
]

自动特征生成：

auto_feature_generation = [
    "feature_interactions",  # 特征交互
    "feature_aggregations", # 特征聚合
    "feature_ratios",       # 特征比值
    "feature_differences",  # 特征差值
    "time_based_features",  # 时间特征
    "text_embeddings",      # 文本嵌入
]

总结

本章详细介绍了 RD-Agent 的核心功能，包括：

命令行界面基础 - 统一的 CLI 设计和监控能力
数据科学代理详解 - 自动化机器学习的完整流程
实际应用场景 - Kaggle 竞赛、医疗预测、通用建模

下一章我们将深入探讨量化金融代理和通用模型代理的高级功能。

快速参考命令：

# 启动数据科学项目
rdagent data_science --competition <project_name>

# 启动监控界面
rdagent ui --port 19899 --data_science True

# 健康检查
rdagent health_check

# 查看帮助
rdagent data_science --help

RD-Agent 教程 - 第一章：基础入门

Henry Lin — Wed, 04 Feb 2026 09:20:59 +0000

RD-Agent 教程 - 第一章：基础入门

1.1 项目概述与核心理念

RD-Agent 简介与背景

RD-Agent (Research & Development Agent) 是由微软研究院开发的开源自动化研发框架，专门用于数据驱动的AI解决方案构建。该项目在业界权威的机器学习工程基准测试 MLE-bench 中获得了第一名的优异成绩，成为目前表现最佳的机器学习工程代理。

核心理念：R&D 框架

RD-Agent 的设计哲学基于两个核心组件：

R (Research): 🔬 研究组件
- 自动提出新的假设和想法
- 分析现有方案的局限性
- 基于领域知识生成创新解决方案
D (Development): 🛠️ 开发组件
- 将研究想法转化为可执行代码
- 实现和验证假设
- 持续优化和改进解决方案

MLE-bench 基准测试表现

Agent	Low (%)	Medium (%)	High (%)	All (%)
R&D-Agent o1-preview	48.18 ± 2.49	8.95 ± 2.36	18.67 ± 2.98	22.4 ± 1.1
R&D-Agent o3(R)+GPT-4.1(D)	51.52 ± 6.21	7.89 ± 3.33	16.67 ± 3.65	22.45 ± 2.45
AIDE o1-preview	34.3 ± 2.4	8.8 ± 1.1	10.0 ± 1.9	16.9 ± 1.1

核心优势与应用价值

1. 自动化数据驱动研发流程

端到端自动化：从数据分析到模型部署的完整流程
智能决策支持：基于数据洞察自动生成解决方案
持续学习能力：从历史经验中学习并不断改进

2. 多代理协作架构

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   研究代理       │    │   开发代理       │    │   评估代理       │
│  Research Agent │◄──►│ Development     │◄──►│ Evaluation      │
│                 │    │ Agent           │    │ Agent           │
├─────────────────┤    ├─────────────────┤    ├─────────────────┤
│• 假设生成        │    │• 代码实现        │    │• 性能评估        │
│• 策略规划        │    │• 测试验证        │    │• 反馈生成        │
│• 知识整合        │    │• 优化改进        │    │• 质量控制        │
└─────────────────┘    └─────────────────┘    └─────────────────┘

3. 持续学习和演进能力

CoSTEER 框架：协作演进策略框架
知识积累：自动构建和维护领域知识库
自适应优化：根据反馈动态调整策略

适用场景与目标用户

🏢 量化投资团队

自动因子挖掘：从海量数据中发现有效因子
策略开发：自动生成和优化交易策略
风险管理：智能风险控制和组合优化
成本效益：相比传统方法，以不到10美元成本获得2倍收益

🔬 数据科学家

特征工程自动化：智能特征生成和选择
模型选择优化：自动化模型架构搜索
实验管理：系统化的实验追踪和对比
管道构建：端到端机器学习管道自动构建

🎓 研究人员和学者

论文复现：自动从论文中提取和实现模型
研究助手：智能文献分析和知识提取
实验设计：科学的实验方案生成和执行
成果整理：自动化研究成果文档化

🏆 Kaggle 竞赛参与者

端到端自动化：从数据探索到模型提交
特征工程：智能特征创建和组合
模型集成：多模型融合策略优化
时间管理：高效的竞赛时间分配

1.2 系统要求与环境准备

硬件和软件要求

操作系统支持

Linux: ✅ 完全支持（Ubuntu 18.04+, CentOS 7+）
Windows: ❌ 当前不支持
macOS: ❌ 当前不支持

重要提示: RD-Agent 目前仅支持 Linux 系统。如果您使用 Windows 或 macOS，建议使用 Docker、WSL2 或虚拟机运行 Linux 环境。

Python 版本要求

Python 3.10: ✅ 完全支持（推荐）
Python 3.11: ✅ 完全支持
Python 3.9: ⚠️ 部分功能可能受限
Python 3.12+: ⚠️ 可能存在兼容性问题

内存和计算资源建议

使用场景	内存要求	CPU 要求	存储空间
轻量级使用	4GB+	2核+	10GB+
中等规模	8GB+	4核+	50GB+
大规模应用	16GB+	8核+	100GB+
生产环境	32GB+	16核+	500GB+

Docker 环境配置

Docker 安装和配置

RD-Agent 强依赖 Docker 来提供隔离的运行环境，确保实验的可重现性。

Ubuntu/Debian 安装 Docker：

# 更新包索引
sudo apt-get update

# 安装必要的包
sudo apt-get install \
    apt-transport-https \
    ca-certificates \
    curl \
    gnupg \
    lsb-release

# 添加 Docker 官方 GPG 密钥
curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg --dearmor -o /usr/share/keyrings/docker-archive-keyring.gpg

# 设置稳定版本仓库
echo \
  "deb [arch=amd64 signed-by=/usr/share/keyrings/docker-archive-keyring.gpg] https://download.docker.com/linux/ubuntu \
  $(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null

# 安装 Docker Engine
sudo apt-get update
sudo apt-get install docker-ce docker-ce-cli containerd.io

CentOS/RHEL 安装 Docker：

# 安装必要的包
sudo yum install -y yum-utils

# 添加 Docker 仓库
sudo yum-config-manager \
    --add-repo \
    https://download.docker.com/linux/centos/docker-ce.repo

# 安装 Docker Engine
sudo yum install docker-ce docker-ce-cli containerd.io

# 启动 Docker 服务
sudo systemctl start docker
sudo systemctl enable docker

无 sudo 权限运行配置

这是必需的配置步骤！ RD-Agent 要求当前用户能够在不使用 sudo 的情况下运行 Docker 命令。

# 创建 docker 用户组（如果不存在）
sudo groupadd docker

# 将当前用户添加到 docker 组
sudo usermod -aG docker $USER

# 重新加载用户组权限（或者重新登录）
newgrp docker

# 验证配置是否成功
docker run hello-world

期望输出：

Hello from Docker!
This message shows that your installation appears to be working correctly.
...

LLM API 准备

支持的 LLM 提供商

RD-Agent 通过 LiteLLM 支持多种大语言模型提供商：

提供商	支持的模型	推荐用途	成本考虑
OpenAI	GPT-4, GPT-4-turbo, GPT-3.5-turbo	通用场景，高质量输出	$$$
Azure OpenAI	GPT-4, GPT-3.5-turbo	企业级应用，合规要求	$$$
DeepSeek	deepseek-chat, deepseek-coder	代码生成，经济实惠	$
Anthropic	Claude-3.x	长文本处理，分析任务	$$
Google	Gemini Pro	多模态需求	$$

API 密钥获取指南

OpenAI API 密钥获取：

访问 OpenAI API 官网
注册/登录账户
前往 "API Keys" 页面
点击 "Create new secret key"
复制并安全保存 API 密钥

Azure OpenAI 配置：

登录 Azure Portal
创建 Azure OpenAI 资源
获取 API 密钥、端点和部署名称
记录 API 版本信息

DeepSeek API 密钥获取：

访问 DeepSeek API 平台
注册账户并完成验证
前往 API 密钥管理页面
生成新的 API 密钥

成本预算建议

使用强度	月成本预算	适用场景
学习试用	$10-50	教程学习，小规模实验
中等使用	$50-200	项目开发，中等规模应用
重度使用	$200-1000	生产环境，大规模应用
企业级	$1000+	企业级部署，持续运行

1.3 安装与基础配置

安装方式选择

用户版本安装（推荐）

对于大多数用户，直接通过 PyPI 安装是最简单的方式：

# 创建 conda 环境（推荐）
conda create -n rdagent python=3.10
conda activate rdagent

# 或使用 venv
python3.10 -m venv rdagent_env
source rdagent_env/bin/activate

# 安装 RD-Agent
pip install rdagent

开发者版本安装

如果您想要体验最新功能或参与开发，可以从源码安装：

# 克隆仓库
git clone https://github.com/microsoft/RD-Agent.git
cd RD-Agent

# 创建开发环境
conda create -n rdagent-dev python=3.10
conda activate rdagent-dev

# 安装开发依赖
make dev

# 或者手动安装
pip install -e .
pip install -r requirements/dev.txt

依赖管理和虚拟环境

为什么需要虚拟环境？

避免包版本冲突
确保环境一致性
便于环境管理和迁移

推荐的环境管理方式：

# 使用 conda（推荐）
conda create -n rdagent python=3.10
conda activate rdagent

# 或使用 pyenv + venv
pyenv install 3.10.12
pyenv local 3.10.12
python -m venv rdagent_env
source rdagent_env/bin/activate

基础环境配置

`.env` 文件配置详解

RD-Agent 使用 .env 文件来管理环境变量配置。在项目根目录创建 .env 文件：

# 在项目目录下创建 .env 文件
touch .env

LiteLLM 统一后端配置

方式一：统一 API 基础配置（推荐新手）

# OpenAI 配置示例
CHAT_MODEL=gpt-4o
EMBEDDING_MODEL=text-embedding-3-small
OPENAI_API_KEY=sk-your-openai-api-key-here
OPENAI_API_BASE=https://api.openai.com/v1

方式二：分离式 API 配置（高级用户）

# 聊天模型配置
CHAT_MODEL=gpt-4o
OPENAI_API_KEY=sk-your-openai-api-key-here
OPENAI_API_BASE=https://api.openai.com/v1

# 嵌入模型配置（使用不同提供商）
EMBEDDING_MODEL=litellm_proxy/BAAI/bge-large-en-v1.5
LITELLM_PROXY_API_KEY=your-siliconflow-api-key
LITELLM_PROXY_API_BASE=https://api.siliconflow.cn/v1

多种 LLM 提供商配置示例

OpenAI 标准配置：

# 基础配置
CHAT_MODEL=gpt-4o
EMBEDDING_MODEL=text-embedding-3-small
OPENAI_API_KEY=sk-proj-xxxxxxxxxxxxxxxxxxxxxxxxxx
OPENAI_API_BASE=https://api.openai.com/v1

# 可选：推理模型特殊配置
# REASONING_THINK_RM=True  # 启用思维过程移除

Azure OpenAI 配置：

# Azure OpenAI 配置
CHAT_MODEL=azure/gpt-4o-deployment-name
EMBEDDING_MODEL=azure/embedding-deployment-name
AZURE_API_KEY=your-azure-api-key
AZURE_API_BASE=https://your-resource.openai.azure.com/
AZURE_API_VERSION=2024-02-15-preview

DeepSeek 经济型配置：

# DeepSeek 聊天模型（经济实惠）
CHAT_MODEL=deepseek/deepseek-chat
DEEPSEEK_API_KEY=sk-your-deepseek-api-key

# SiliconFlow 嵌入模型（DeepSeek 无嵌入模型）
EMBEDDING_MODEL=litellm_proxy/BAAI/bge-m3
LITELLM_PROXY_API_KEY=sk-your-siliconflow-api-key
LITELLM_PROXY_API_BASE=https://api.siliconflow.cn/v1

# 推理模型配置
REASONING_THINK_RM=True

混合配置（推荐高级用户）：

# 高性能聊天模型
CHAT_MODEL=gpt-4o
OPENAI_API_KEY=sk-your-openai-api-key

# 经济型嵌入模型
EMBEDDING_MODEL=litellm_proxy/BAAI/bge-large-en-v1.5
LITELLM_PROXY_API_KEY=sk-your-siliconflow-api-key
LITELLM_PROXY_API_BASE=https://api.siliconflow.cn/v1

# 其他配置
REASONING_THINK_RM=False

健康检查与验证

`rdagent health_check` 使用

安装完成后，首先进行系统健康检查：

# 完整健康检查
rdagent health_check

# 跳过环境变量检查
rdagent health_check --no-check-env

# 跳过 Docker 检查
rdagent health_check --no-check-docker

# 只检查端口占用
rdagent health_check --no-check-env --no-check-docker

期望的成功输出：

✅ Docker is installed and working correctly
✅ Port 19899 is available
✅ Environment variables are configured correctly
✅ LLM API connection successful
✅ All systems check passed!

常见配置问题排查

问题 1：Docker 权限错误

❌ Error: Got permission denied while trying to connect to the Docker daemon socket

解决方案：

# 确保用户在 docker 组中
sudo usermod -aG docker $USER
newgrp docker
# 重新运行健康检查
rdagent health_check

问题 2：API 密钥无效

❌ Error: Invalid API key provided

解决方案：

# 检查 .env 文件中的 API 密钥
cat .env | grep API_KEY
# 确保密钥正确且有效
# 重新配置 API 密钥

问题 3：端口被占用

❌ Error: Port 19899 is already in use

解决方案：

# 查看端口占用情况
netstat -tlnp | grep 19899
# 结束占用进程或更换端口
rdagent ui --port 19900

问题 4：网络连接问题

❌ Error: Failed to connect to API endpoint

解决方案：

# 检查网络连接
curl -I https://api.openai.com/v1/models
# 配置代理（如需要）
export https_proxy=http://proxy.company.com:8080
export http_proxy=http://proxy.company.com:8080

环境验证清单

在开始使用 RD-Agent 之前，请确认以下检查项：

[ ] 操作系统: Linux 环境
[ ] Python 版本: 3.10 或 3.11
[ ] 虚拟环境: 已激活专用环境
[ ] Docker: 安装并可无 sudo 运行
[ ] 网络连接: 能访问 LLM API 端点
[ ] API 密钥: 有效且有足够余额
[ ] 存储空间: 至少 10GB 可用空间
[ ] 端口: 19899 端口可用（或指定其他端口）
[ ] 权限: 当前用户有读写项目目录权限

至此，您已完成 RD-Agent 的基础安装和配置。下一章我们将详细介绍 RD-Agent 的核心功能和使用方法。

快速验证安装

完成所有配置后，让我们进行一个快速的功能验证：

# 激活环境
conda activate rdagent

# 验证安装
python -c "import rdagent; print('RD-Agent installed successfully!')"

# 运行健康检查
rdagent health_check

# 查看可用命令
rdagent --help

如果所有步骤都正常完成，您现在已经准备好开始使用 RD-Agent 的强大功能了！

Alpha158和Alpha360因子计算公式文档

Henry Lin — Mon, 02 Feb 2026 07:52:12 +0000

Alpha158和Alpha360因子计算公式文档

概述

本文档详细描述了Qlib量化投资平台中Alpha158和Alpha360数据集的因子计算公式。这些因子集被广泛用于股票预测模型的训练和评估。

Alpha158因子集

Alpha158包含158个技术分析指标，分为四大类：K线特征、价格特征、成交量特征和滚动窗口技术指标。

1. K线基础特征 (9个)

K线基础特征基于单日OHLC数据计算，反映价格形态特征：

因子名称	计算公式	含义
KMID	`($close-$open)/$open`	收盘与开盘价差比例，反映日内涨跌幅
KLEN	`($high-$low)/$open`	最高最低价差比例，反映日内波动幅度
KMID2	`($close-$open)/($high-$low+1e-12)`	收开差与振幅比，反映实体占比
KUP	`($high-Greater($open, $close))/$open`	上影线比例，反映上方压力
KUP2	`($high-Greater($open, $close))/($high-$low+1e-12)`	上影线占总振幅比例
KLOW	`(Less($open, $close)-$low)/$open`	下影线比例，反映下方支撑
KLOW2	`(Less($open, $close)-$low)/($high-$low+1e-12)`	下影线占总振幅比例
KSFT	`(2*$close-$high-$low)/$open`	收盘价相对于中点位置
KSFT2	`(2*$close-$high-$low)/($high-$low+1e-12)`	收盘价在振幅中的相对位置

2. 价格特征 (4个)

基础价格特征，标准化为相对当前收盘价的比值：

因子名称	计算公式	含义
OPEN0	`$open/$close`	当日开盘价相对收盘价比值
HIGH0	`$high/$close`	当日最高价相对收盘价比值
LOW0	`$low/$close`	当日最低价相对收盘价比值
VWAP0	`$vwap/$close`	当日成交均价相对收盘价比值

3. 滚动窗口技术指标 (145个)

使用5,10,20,30,60天滚动窗口计算的技术指标，每个指标生成5个特征，共29类指标。

3.1 价格趋势与动量类

因子类别	计算公式	含义	参考链接
ROC	`Ref($close, d)/$close`	d日前价格变化率	Rate of Change
MA	`Mean($close, d)/$close`	d日简单移动平均	Moving Average
STD	`Std($close, d)/$close`	d日价格标准差，反映波动性	-
BETA	`Slope($close, d)/$close`	d日价格趋势斜率	-
RSQR	`Rsquare($close, d)`	d日线性回归R²值，反映趋势线性度	-
RESI	`Resi($close, d)/$close`	d日线性回归残差	-

3.2 价格位置类

因子类别	计算公式	含义
MAX	`Max($high, d)/$close`	d日内最高价
MIN	`Min($low, d)/$close`	d日内最低价
QTLU	`Quantile($close, d, 0.8)/$close`	d日80%分位数
QTLD	`Quantile($close, d, 0.2)/$close`	d日20%分位数
RANK	`Rank($close, d)`	当前价格在d日内排名百分位
RSV	`($close-Min($low, d))/(Max($high, d)-Min($low, d)+1e-12)`	相对强弱值，KDJ指标基础

3.3 时间序列位置类

因子类别	计算公式	含义	参考链接
IMAX	`IdxMax($high, d)/d`	最高价出现的相对位置	Aroon Indicator
IMIN	`IdxMin($low, d)/d`	最低价出现的相对位置	Aroon Indicator
IMXD	`(IdxMax($high, d)-IdxMin($low, d))/d`	高低点时间差，反映下跌动量	-

3.4 价格-成交量关联类

因子类别	计算公式	含义
CORR	`Corr($close, Log($volume+1), d)`	价格与成交量相关性
CORD	`Corr($close/Ref($close,1), Log($volume/Ref($volume, 1)+1), d)`	价格变化与成交量变化相关性

3.5 涨跌统计类

因子类别	计算公式	含义
CNTP	`Mean($close>Ref($close, 1), d)`	d日内上涨天数比例
CNTN	`Mean($close<Ref($close, 1), d)`	d日内下跌天数比例
CNTD	`CNTP - CNTN`	涨跌天数差

3.6 RSI类指标

因子类别	计算公式	含义	参考链接
SUMP	`Sum(Greater($close-Ref($close, 1), 0), d)/(Sum(Abs($close-Ref($close, 1)), d)+1e-12)`	总收益占总变动比例	RSI Indicator
SUMN	`Sum(Greater(Ref($close, 1)-$close, 0), d)/(Sum(Abs($close-Ref($close, 1)), d)+1e-12)`	总亏损占总变动比例	RSI Indicator
SUMD	`SUMP - SUMN`	收益亏损差值比，类RSI指标	RSI Indicator

3.7 成交量技术指标

因子类别	计算公式	含义	参考链接
VMA	`Mean($volume, d)/($volume+1e-12)`	成交量移动平均	Volume Moving Average
VSTD	`Std($volume, d)/($volume+1e-12)`	成交量标准差	-
WVMA	`Std(Abs($close/Ref($close, 1)-1)$volume, d)/(Mean(Abs($close/Ref($close, 1)-1)$volume, d)+1e-12)`	成交量加权价格变动波动率	-
VSUMP	`Sum(Greater($volume-Ref($volume, 1), 0), d)/(Sum(Abs($volume-Ref($volume, 1)), d)+1e-12)`	成交量增加占比	-
VSUMN	`Sum(Greater(Ref($volume, 1)-$volume, 0), d)/(Sum(Abs($volume-Ref($volume, 1)), d)+1e-12)`	成交量减少占比	-
VSUMD	`VSUMP - VSUMN`	成交量变化差值比	-

滚动窗口说明

所有滚动指标都使用以下窗口大小：[5, 10, 20, 30, 60]天

因此每个滚动指标会生成5个特征，例如：

ROC5, ROC10, ROC20, ROC30, ROC60
MA5, MA10, MA20, MA30, MA60
STD5, STD10, STD20, STD30, STD60
...以此类推

Alpha360因子集

Alpha360提供360个原始价格特征，包含过去60天的完整OHLCV数据序列。

设计理念

原始数据保留：提供原始价格数据而非工程化特征
统一标准化：所有价格通过最新收盘价标准化，成交量通过最新成交量标准化
时间序列完整性：保持60天完整时间序列，便于深度学习模型使用

因子构成

Alpha360包含6类因子，每类60个特征，共360个：

1. 收盘价序列 (60个)

CLOSE59, CLOSE58, CLOSE57, ..., CLOSE2, CLOSE1, CLOSE0

计算公式：

CLOSE59: Ref($close, 59)/$close (59天前收盘价/当前收盘价)
CLOSE58: Ref($close, 58)/$close (58天前收盘价/当前收盘价)
...
CLOSE1: Ref($close, 1)/$close (1天前收盘价/当前收盘价)
CLOSE0: $close/$close = 1 (当前收盘价，标准化后恒为1)

2. 开盘价序列 (60个)

OPEN59, OPEN58, OPEN57, ..., OPEN2, OPEN1, OPEN0

计算公式：Ref($open, i)/$close (i从59到0)

3. 最高价序列 (60个)

HIGH59, HIGH58, HIGH57, ..., HIGH2, HIGH1, HIGH0

计算公式：Ref($high, i)/$close (i从59到0)

4. 最低价序列 (60个)

LOW59, LOW58, LOW57, ..., LOW2, LOW1, LOW0

计算公式：Ref($low, i)/$close (i从59到0)

5. 成交均价序列 (60个)

VWAP59, VWAP58, VWAP57, ..., VWAP2, VWAP1, VWAP0

计算公式：Ref($vwap, i)/$close (i从59到0)

6. 成交量序列 (60个)

VOLUME59, VOLUME58, VOLUME57, ..., VOLUME2, VOLUME1, VOLUME0

计算公式：Ref($volume, i)/($volume+1e-12) (i从59到0)

标准化说明

价格标准化：所有价格(收盘、开盘、最高、最低、VWAP)都除以当前收盘价
成交量标准化：所有成交量都除以当前成交量(加1e-12防止除零)
标准化结果：CLOSE0恒为1，VOLUME0恒为1(进一步标准化后可能为0)

标签定义

两个因子集都使用相同的标签定义：

["Ref($close, -2)/Ref($close, -1) - 1"], ["LABEL0"]

含义：预测未来1天的收益率

分子：Ref($close, -2) = t+2日的收盘价
分母：Ref($close, -1) = t+1日的收盘价
标签：(t+2日收盘价 / t+1日收盘价) - 1

使用场景对比

Alpha158适用场景

传统机器学习模型：LightGBM、XGBoost、线性回归等
特征工程完备：包含丰富的技术分析指标
解释性强：每个特征都有明确的金融含义
计算效率高：预计算的技术指标，模型训练快速

Alpha360适用场景

深度学习模型：LSTM、GRU、Transformer等时间序列模型
序列建模：保留完整的时间序列信息
自动特征提取：模型自动学习价格模式
端到端学习：从原始价格数据直接学习预测

数据质量与预处理

缺失值处理

Alpha158中包含除法运算，分母加入1e-12防止除零
成交量相关计算使用($volume+1e-12)处理零成交量情况
建议训练前进行NaN值检查和填充

异常值处理

价格跳空、停牌可能导致极值，建议进行异常值检测
成交量为0的情况已通过1e-12处理
建议对特征进行winsorize处理，削减极端值影响

标准化建议

Alpha158：建议进一步Z-score标准化
Alpha360：已经过相对标准化，但建议根据模型需求进一步处理

实现代码位置

Alpha158实现：qlib/contrib/data/loader.py:72-310 (Alpha158DL.get_feature_config())
Alpha360实现：qlib/contrib/data/loader.py:16-58 (Alpha360DL.get_feature_config())
数据处理器：qlib/contrib/data/handler.py:48-158 (Alpha158和Alpha360类)

Crypto Forem: Henry Lin

Hermes Agent 项目功能与通用使用场景分析

Hermes Agent 项目功能与通用使用场景分析

1. 项目定位

2. 项目主干结构

3. 核心功能拆解

3.1 Agent 会话循环

3.2 工具注册与工具集

3.3 终端与多执行环境

3.4 CLI 交互系统

3.5 消息网关

3.6 会话持久化与检索

3.7 记忆与技能系统

3.8 上下文压缩与长任务支持

3.9 子代理与程序化工具调用

3.10 定时任务与后台自动化

3.11 MCP 与外部系统扩展

3.12 批处理与研究型能力

4. 项目的通用能力画像

5. 典型使用场景

5.1 个人长期助手

5.2 开发与调试助手

5.3 研究与资料整理助手

5.4 运维与巡检助手

5.5 团队协作 Agent

5.6 文档与知识管理助手

5.7 自动化报告系统

5.8 多系统统一入口

6. 项目的工程价值

6.1 它不是玩具式 agent

6.2 它强调统一入口

6.3 它强调可扩展性

6.4 它强调长期记忆和复用

7. 适合解决的问题类型

8. 不适合解决的问题类型

8.1 不适合极低延迟场景

8.2 不适合强确定性核心逻辑

8.3 不适合无边界的高权限自动执行

9. 最适合的落地方式

9.1 本地个人工作台

9.2 服务器上的长期运行 Agent

9.3 组织内部的统一智能接口

10. 最终评价

第1章 什么是 QuantConnect/Lean

第1章 什么是 Lean？

1.1 项目简介

核心功能

支持的编程语言

1.2 Lean vs Lean CLI

Lean Engine（本项目）

Lean CLI（推荐用于策略开发）

1.3 核心特性

事件驱动架构

多资产类别支持

模块化设计

1.4 架构概览

1.5 本地与云端混合开发

1.6 示例：第一个 Lean 算法

# RD-Agent Tutorial - Chapter 3: Advanced Usage Techniques

RD-Agent Tutorial - Chapter 3: Advanced Usage Techniques

3.1 Quantitative Finance Agent Deep Dive

Quantitative Finance Scenario Architecture

Core Architecture Design

Qlib Framework Integration

Factor Mining and Optimization

rdagent fin_factor Explained

Factor Generation Strategies

Factor Effectiveness Evaluation

Model Evolution and Optimization

rdagent fin_model Explained

Model Architecture Search

Hyperparameter Optimization

Factor-Model Joint Optimization

rdagent fin_quant Explained

Multi-Objective Optimization Algorithms

Collaborative Evolution Mechanism

Financial Report Analysis

rdagent fin_factor_report Usage

Report Parsing and Feature Extraction

Sentiment Analysis and Signal Mining

第1章什么是 QuantConnect/Lean

第1章什么是 Lean？

`rdagent fin_factor` Explained

`rdagent fin_model` Explained

`rdagent fin_quant` Explained

`rdagent fin_factor_report` Usage