WorldQuant Alpha因子研究多智能体系统

6个专业AI智能体组成的自动化流水线，将研报自动加工成可验证的Alpha因子公式

---

🎯 核心架构

一个由7个专业AI智能体组成的流水线，像工厂一样将研报自动加工成可验证的Alpha因子公式。

智能体团队

| 智能体 | 角色 | 代码文件 | 职责 | |--------|------|----------|------| | 🕵️ 情报员 | 研报爬取 | research_crawler.py | 搜刮各类研报和策略文章，产出清洗好的文本材料 | | 📚 合规官 | 数据管理 | data_manager.py | 掌握WorldQuant所有数据字段和操作符规范，产出权威的"语法词典" | | 🔄 翻译官 | 策略分析 | strategy_analyst.py | 把研报文字转化成量化逻辑，产出清晰的量化策略描述 | | 💻 程序员 | Alpha工程师 | alpha_engineer.py | 把量化逻辑变成WorldQuant公式，产出合规的Alpha因子代码，支持网格搜索和公式优化 | | 🧪 测试员 | 回测师 | backtester.py | 在WorldQuant平台测试公式效果，产出回测结果报告 | | 📊 评审官 | 评审员 | critic.py | 评估回测结果，提改进建议，产出质量评分+优化建议 |

工作流程

【输入】研报/想法
    ↓
情报员 → 收集资料
    ↓
合规官 → 提供规范
    ↓
翻译官 → 理解逻辑
    ↓
程序员 → 编写公式 + 参数搜索（网格搜索）
    ↓
测试员 → 测试效果
    ↓
评审官 → 评估质量
    ↓
是否满足条件？
    ↓ 否 → 优化建议 → 程序员 → 优化公式 → 返回测试
    ↓ 是
【输出】有效的Alpha因子 + 完整报告

---

✨ 关键特点

• 🤖 全自动：从想法到验证无人干预
• 📝 守规矩：严格遵守WorldQuant语法
• 💾 有记忆：记录所有步骤，可追溯
• 🎓 能学习：失败案例反馈优化
• 🔧 可扩展：随时增加新数据源、新策略
• 🎯 参数优化：网格搜索自动寻找最优settings
• 📊 智能评估：AI深度分析回测结果
• 🔄 迭代优化：自动优化公式直到满足条件
• 📈 历史追踪：完整记录优化历史，支持回滚

---

💡 系统价值

• 🌱 对新手：帮你把想法快速变成可测试的公式
• 🏆 对老手：批量处理研报，发现新策略
• 👥 对团队：标准化研究流程，积累知识库

本质：一个24小时不休息、不出错、严格守规的数字量化研究员团队。

---

🚀 快速开始

环境要求

• Python 3.11+
• Conda (推荐)

安装步骤

1. 克隆项目

git clone <repository-url>
cd wq自动化

1. 创建Conda环境

conda env create -f environment.yml
conda activate alpha_research

或者手动安装依赖：

pip install -r requirements.txt

1. 配置文件 编辑 config.yaml，设置API密钥：

worldquant:
  api_key: "YOUR_API_KEY"

ai:
  api_key: "YOUR_DEEPSEEK_KEY"

启动项目（重要）

由于CrewAI框架需要OpenAI兼容的环境变量，请使用提供的启动脚本：

Windows：

run_with_env.bat python main.py

Linux/Mac：

chmod +x run_with_env.sh
./run_with_env.sh python main.py

启动脚本会自动设置：

• OPENAI_API_KEY：从config.yaml读取的DeepSeek密钥
• OPENAI_API_BASE：DeepSeek API地址
• PYTHONIOENCODING：UTF-8编码（解决中文乱码）

注意：

• 不要直接运行 python main.py，否则会报错
• 如果遇到中文乱码，使用启动脚本即可解决

---

📁 项目结构

wq自动化/
├── agents/                      # 7个智能体实现
│   ├── base_agent.py           # 基础智能体类
│   ├── research_crawler.py     # 🕵️ 情报员：研报爬取
│   ├── data_manager.py         # 📚 合规官：数据管理
│   ├── strategy_analyst.py     # 🔄 翻译官：策略分析
│   ├── alpha_engineer.py       # 💻 程序员：公式生成
│   ├── backtester.py           # 🧪 测试员：回测执行
│   ├── critic.py               # 📊 评审官：质量评估
│   └── grid_search_agent.py    # 🔍 优化师：网格搜索
├── tools/                       # 工具函数
│   ├── ai_client.py            # AI客户端
│   ├── web_scraper.py          # 网页爬虫
│   └── worldquant_client.py    # WorldQuant客户端
├── utils/                       # 工具函数
│   ├── config_loader.py        # 配置加载
│   ├── logger.py               # 日志配置
│   └── optimization_history.py # 优化历史记录
├── schemas/                     # 数据规范
│   └── __init__.py
├── tests/                       # 测试文件
│   ├── test_alpha_engineer_fixed.py   # Alpha工程师测试
│   ├── test_backtester_complete.py    # 回测师测试
│   ├── test_critic.py                 # 评论员测试
│   └── test_full_workflow.py          # 完整工作流测试
├── examples/                    # 示例代码
│   ├── agent_workflow_example.py      # 智能体交互示例
│   ├── example_grid_search.py         # 网格搜索示例
│   └── example_optimization_history.py # 优化历史示例
├── scripts/                     # 辅助脚本
│   ├── fetch_data_specs.py      # 获取数据规范
│   ├── fetch_operators.py       # 获取操作符
│   ├── generate_field_mapping.py # 生成字段映射
│   ├── monitor_progress.py      # 监控进度
│   └── continue_backtest.py     # 继续回测
├── logs/                        # 日志文件目录
├── results/                     # 结果存储目录
├── __pycache__/                 # Python缓存
├── crew.py                      # CrewAI协调器
├── main.py                      # 主程序入口
├── batch_backtest.py            # 批量回测脚本
├── machine_lib.py               # 参考代码
├── config.yaml                  # 配置文件
├── requirements.txt             # 依赖包列表
├── environment.yml              # Conda环境配置
├── worldquant_specs.json        # WorldQuant数据规范
├── worldquant_operators.json    # WorldQuant操作符
├── run_with_env.bat             # Windows启动脚本
├── run_with_env.sh              # Linux/Mac启动脚本
├── README.md                    # 项目说明
└── .gitignore                   # Git忽略规则

---

🔧 配置说明

config.yaml

# WorldQuant平台配置
worldquant:
  api_key: "YOUR_API_KEY"
  base_url: "https://api.worldquantbrain.com"
  timeout: 30

# AI模型配置
ai:
  api_key: "YOUR_DEEPSEEK_KEY"
  model: "deepseek-chat"
  temperature: 0.1

# 日志配置
logging:
  level: "INFO"
  file: "logs/alpha_research.log"
  max_size_mb: 10
  backup_count: 5

# 研报爬取配置
crawler:
  timeout: 30
  retry_times: 3

# 回测配置
backtest:
  default_universe: "all"
  start_date: "2020-01-01"
  end_date: "2024-12-31"
  frequency: "daily"
  benchmark: "000300.SH"

  # WorldQuant提交条件标准
  submission_criteria:
    fitness: 1.0                    # 适配度 ≥1.0
    sharpe_ratio: 1.25              # 夏普比率 ≥1.25
    turnover_min: 0.01              # 换手率 ≥1%
    turnover_max: 0.70              # 换手率 ≤70%
    max_weight: 0.10                # 最大股票权重 <10%
    correlation_threshold: 0.7      # 与现有因子相关性 <0.7

  # WorldQuant回测settings配置
  settings:
    instrumentType: "EQUITY"        # 证券类型
    region: "USA"                   # 区域：USA/CHN/HKG/JPN/EUR/ASI/GLB等
    universe: "TOP3000"             # 股票池：TOP3000/TOP1000/TOP500/TOP200/ALL等
    delay: 1                        # 延迟：1(D1,使用昨日数据) 或 0(D0,使用当日估算)
    decay: 20                       # 衰减：正整数，如4, 5, 10, 20等
    neutralization: "SECTOR"        # 中性化：MARKET/INDUSTRY/SUBINDUSTRY/SECTOR/NONE
    truncation: 0.08                # 截断：小数，如0.01(1%), 0.08(8%)
    pasteurization: "ON"            # 巴氏处理：ON/OFF
    testPeriod: "P2Y"               # 测试周期：P0Y(样本内)/P1Y/P2Y/P3Y(样本外)
    unitHandling: "VERIFY"          # 单位处理：VERIFY
    nanHandling: "ON"               # 空值处理：ON/OFF
    language: "FASTEXPR"            # 表达式语言
    visualization: false            # 可视化

  # 网格搜索参数范围
  grid_search:
    enabled: true
    decay: [5, 10, 15, 20, 25, 30]
    neutralization: ["MARKET", "INDUSTRY", "SUBINDUSTRY", "SECTOR"]
    truncation: [0.01, 0.05, 0.08, 0.10]
    max_iterations: 10
    improvement_threshold: 0.05

---

🔄 批量回测功能

回测师智能体支持批量回测功能，可以高效地处理大量Alpha公式。

功能特点

• 分组回测：将公式分成N个一组（默认每组3条），并发提交到WorldQuant平台
• 登录重试：自动检测会话过期（4小时后），自动重新登录
• 进度保存：实时保存回测进度到 batch_backtest_progress.json
• 断点续传：中断后可从任意组继续执行
• 并发处理：自动处理429并发限制错误，等待30秒后重试

使用方法

基本批量回测

# 使用默认参数（每组3条）
python batch_backtest.py

# 自定义参数
python batch_backtest.py --group-size 3 --formula-file your_formulas.json

断点续传

如果回测过程中断，可以从指定组继续：

# 从第5组继续
python batch_backtest.py --start-group 5 --group-size 3

参数说明

| 参数 | 默认值 | 说明 | |------|--------|------| | --formula-file | test_50_reports_with_new_engineer.json | 公式文件路径 | | --group-size | 3 | 每组公式数量（建议3） | | --start-group | 0 | 起始组索引（用于断点续传） | | --decay | 20 | 衰减参数 | | --region | USA | 区域 | | --universe | TOP3000 | 股票池 | | --neutralization | SECTOR | 中性化方式 |

进度文件

进度会自动保存到 batch_backtest_progress.json，包含：

• 当前组索引
• 已完成的组列表
• 回测结果

结果文件

结果会保存到 batch_backtest_results_YYYYMMDD_HHMMSS.json，包含每组回测的详细信息。

代码示例

from agents.backtester import BacktesterAgent

# 创建回测师智能体
backtester = BacktesterAgent()

# 运行批量回测
results = backtester.run_batch_backtest(
    formula_file='test_50_reports_with_new_engineer.json',
    group_size=3,  # 每组3条公式
    start_group=0  # 从第0组开始
)

# 查看结果
for result in results:
    print(f"组 {result['group_index']}: {result['status']}")

注意事项

1. 并发限制：WorldQuant平台对并发回测有限制，建议每组3条公式
2. 会话过期：平台会话4小时后自动过期，系统会自动重新登录
3. 组间延迟：每组完成后会等待10秒再处理下一组，避免触发限制
4. 进度保存：每组完成后立即保存进度，确保不丢失数据

---

🔍 网格搜索功能

网格搜索智能体支持自动寻找最优的回测参数组合，提升Alpha因子的表现。

功能特点

• 参数空间探索：系统性搜索decay、neutralization、truncation等参数的组合
• 自动优化：自动执行回测并评估结果，选择最优参数
• 自适应搜索：支持多轮搜索，根据结果动态调整搜索范围
• 参数敏感性分析：分析单个参数对因子表现的影响
• 历史记录：所有搜索过程自动记录，支持回滚和对比

使用方法

基础网格搜索

from agents.grid_search_agent import GridSearchAgent

# 初始化网格搜索智能体
grid_search = GridSearchAgent()

# 执行网格搜索
result = grid_search.search_best_settings(
    formula="rank(ts_mean(returns, 20))",
    original_logic="价格动量策略",
    strategy_key="momentum_strategy",
    custom_ranges={
        'decay': [10, 15, 20, 25, 30],
        'neutralization': ['MARKET', 'INDUSTRY', 'SECTOR'],
        'truncation': [0.01, 0.05, 0.08]
    },
    max_iterations=10
)

# 查看结果
if result["success"]:
    print(f"最优settings: {result['best_settings']}")
    print(f"最优适配度: {result['best_fitness_score']:.2%}")

自适应搜索

# 执行自适应搜索（多轮优化）
adaptive_result = grid_search.adaptive_search(
    formula="rank(ts_mean(returns, 20))",
    original_logic="价格动量策略",
    strategy_key="momentum_strategy_adaptive",
    max_iterations=20
)

参数敏感性分析

# 分析decay参数的敏感性
sensitivity = grid_search.analyze_parameter_sensitivity(
    formula="rank(ts_mean(returns, 20))",
    original_logic="价格动量策略",
    base_settings={...},
    parameter='decay',
    values=[5, 10, 15, 20, 25, 30, 35, 40]
)

配置说明

在config.yaml中配置网格搜索参数：

backtest:
  grid_search:
    enabled: true
    decay: [5, 10, 15, 20, 25, 30]
    neutralization: ["MARKET", "INDUSTRY", "SUBINDUSTRY", "SECTOR"]
    truncation: [0.01, 0.05, 0.08, 0.10]
    max_iterations: 10
    improvement_threshold: 0.05

参数说明

| 参数 | 说明 | |------|------| | decay | 衰减参数，影响持仓时间和换手率 | | neutralization | 中性化方式，如MARKET、INDUSTRY、SECTOR | | truncation | 截断阈值，控制单只股票最大权重 | | max_iterations | 最大迭代次数 | | improvement_threshold | 改进阈值，用于提前终止搜索 |

运行示例

# 运行网格搜索示例
python examples/example_grid_search.py

---

📊 优化历史记录

优化历史记录管理器支持记录、查询和回滚Alpha因子的优化历史。

功能特点

• 完整记录：记录每次优化的详细信息（公式、settings、回测结果、评估结果）
• 适配度计算：自动计算综合评分，评估因子表现
• 最佳查询：快速查询最佳公式
• 趋势分析：查看优化趋势，了解改进过程
• 版本回滚：支持回滚到任意历史版本
• 报告导出：导出优化报告，便于分析

使用方法

记录优化历史

from utils.optimization_history import OptimizationHistory

# 初始化优化历史记录管理器
history = OptimizationHistory("optimization_history.json")

# 开始新的优化
optimization_id = history.start_optimization(
    strategy_key="my_strategy",
    original_logic="价格动量策略",
    original_formula="rank(ts_mean(returns, 20))",
    fields_used=["returns"]
)

# 记录优化迭代
history.record_iteration(
    strategy_key="my_strategy",
    optimization_id=optimization_id,
    iteration=1,
    formula="rank(ts_mean(returns, 20))",
    formula_name="alpha_v1",
    settings={"decay": 20, "neutralization": "SECTOR", "truncation": 0.08},
    backtest_result={"success": True, "performance": {...}},
    evaluation={"overall_rating": "良好", "score": 65, "is_valid": False},
    optimization_reason="初始公式"
)

查询最佳公式

# 查询最佳公式
best = history.get_best_formula("my_strategy", optimization_id)
print(f"最佳公式: {best['formula_name']}")
print(f"适配度: {best['fitness_score']:.2f}")

查看优化趋势

# 查看优化趋势
trend = history.get_optimization_trend("my_strategy", optimization_id)
for t in trend:
    print(f"迭代{t['iteration']}: 适配度={t['fitness_score']:.2f}, 夏普比率={t['sharpe_ratio']:.2f}")

回滚到指定版本

# 回滚到第2个版本
version_2 = history.rollback_to_version("my_strategy", optimization_id, 2)

导出优化报告

# 导出优化报告
report = history.export_report("optimization_report.json")

运行示例

# 运行优化历史示例
python examples/example_optimization_history.py

---

📊 使用示例

示例1：从研报文本生成Alpha因子

from crew import AlphaResearchCrew

# 初始化团队
crew = AlphaResearchCrew()

# 输入研报
report = """
根据近期市场表现，小市值股票在牛市中表现优于大市值股票。
特别是市值在50亿以下、市盈率低于30倍的股票，未来一个月有较高超额收益。
"""

# 执行完整流程
result = crew.execute_full_workflow({
    "text": report,
    "source": "manual_input"
})

# 查看结果
if result["success"]:
    print(f"Alpha公式: {result['summary']['formula']}")
    print(f"年化收益: {result['summary']['annual_return']:.2%}")
    print(f"夏普比率: {result['summary']['sharpe_ratio']:.2f}")
    print(f"评级: {result['summary']['overall_rating']}")

示例2：从文件加载研报

result = crew.execute_full_workflow({
    "file_path": "reports/strategy_001.txt"
})

---

📈 输出示例

{
  "success": true,
  "summary": {
    "formula": "rank(market_cap) * -1 * (pe_ratio < 30) * (close > ts_mean(close, 20))",
    "annual_return": 0.1587,
    "sharpe_ratio": 1.82,
    "max_drawdown": 0.1243,
    "overall_rating": "良好",
    "recommendation": "继续使用"
  },
  "workflow": {
    "step1_crawl": { "success": true, "report_text": "..." },
    "step2_analysis": { "involved_fields": ["market_cap", "pe_ratio", "close"] },
    "step3_formula": { "formula": "..." },
    "step4_validation": { "is_valid": true },
    "step5_backtest": { "performance": {...} },
    "step6_evaluation": { "overall_rating": "良好" }
  }
}

---

🛠️ 技术栈

• 核心框架: CrewAI (多智能体协作)
• AI模型: DeepSeek API (兼容OpenAI格式)
• 数据处理: Pandas, NumPy
• 配置管理: PyYAML, python-dotenv
• 日志系统: Python logging with rotation
• 数据验证: Pydantic

---

📝 开发计划

• [x] 阶段一：基础框架搭建
• [x] 阶段二：6个智能体实现
• [x] 阶段三：智能体集成
• [x] 阶段四：真实API对接
• [x] 阶段五：网格搜索功能
• [x] 阶段六：优化历史记录
• [x] 阶段七：Settings参数配置化
• [x] 阶段八：AI评估优化
• [x] 阶段九：公式自动优化
• [ ] 阶段十：Web界面开发
• [ ] 阶段十一：性能优化
• [ ] 阶段十二：更多数据源集成

---

🤝 贡献指南

欢迎提交 Issue 和 Pull Request！

1. Fork 本项目
2. 创建特性分支 (git checkout -b feature/AmazingFeature)
3. 提交更改 (git commit -m 'Add some AmazingFeature')
4. 推送到分支 (git push origin feature/AmazingFeature)
5. 开启 Pull Request

---

📄 许可证

本项目采用 MIT 许可证 - 详见 LICENSE 文件

---

👥 联系方式

• 项目主页: [GitHub Repository]
• 问题反馈: [Issues]

---

🙏 致谢

• CrewAI - 强大的多智能体协作框架
• DeepSeek - AI模型支持
• WorldQuant - Alpha因子研究平台

---

⭐ 如果这个项目对你有帮助，请给个 Star！