AITrade - AI量化交易系统

基于ClassRoom课程的插件化、可扩展量化交易平台。

项目状态

阶段	状态	完成时间
Phase 1: 用户认证完善 + 回测引擎完善	✅ 已完成	2026-04-26
Phase 2: analysis分析模块	✅ 已完成	2026-04-26
Phase 3: LangGraph工作流 + Kris风控 + RL策略	✅ 已完成	2026-04-26
Phase 4: V3.0 深度学习预测	✅ 已完成	2026-04-26

Phase 4 完成内容（部分）

LSTM股价预测模型 - ✅ 已完成

✅ LSTMPredictor（PyTorch LSTM + 线性回归降级）
✅ 数据准备与归一化
✅ 训练循环（MSE Loss + Adam优化器）
✅ 多步预测（自回归预测）
✅ 模型保存/加载
✅ 单元测试：6个测试用例全部通过

Transformer时间序列预测 - ✅ 已完成

✅ TransformerPredictor（PyTorch Transformer + 移动平均降级）
✅ 位置编码（Positional Encoding）
✅ 多头自注意力机制
✅ 训练循环（MSE Loss + Adam优化器）
✅ 多步预测（自回归预测）
✅ 模型保存/加载
✅ 单元测试：6个测试用例全部通过

Phase 3 完成内容

LangGraph交易团队工作流 - ✅ 已完成

✅ State定义（TradingState TypedDict）
✅ Charles节点（投研情报官）
✅ Zoe节点（信号官）
✅ Kris节点（风控官）
✅ Human节点（人在回路）
✅ Ethan节点（交易员）
✅ 图编排（StateGraph + Conditional Edge）
✅ 重试/熔断机制

Phase 2 完成内容

✅ 技术指标：MACD、RSI、KDJ、布林带、均线系统
✅ K线形态识别：十字星、锤子线、吞没形态
✅ 趋势分析：趋势判断、支撑/压力位计算
✅ API接口：7个技术分析API
✅ 单元测试：16个测试用例全部通过

Phase 3 完成内容（部分）

Kris风控体系（8条核心规则）- ✅ 已完成

事前风控（Pre-Trade）:

✅ MaxOrderAmountRule - 单笔金额上限
✅ ConcentrationLimitRule - 集中度限制
✅ ATRPositionRule - ATR动态仓位控制
✅ EventKeywordRule - 新闻/研报风险词扫描

事中风控（Intra-Trade）:

✅ MaxDrawdownRule - 最大回撤
✅ MaxDailyLossRule - 最大日亏损
✅ MaxTradeCountRule - 交易频率限制

事后风控（Post-Trade）:

✅ CircuitBreakerRule - 熔断机制（连续亏损/冷却期）
✅ MacroGateRule - 宏观门控（VIX/失业率/通胀率）
✅ 单元测试：23个测试用例全部通过

强化学习交易策略 - ✅ 已完成

✅ DQN智能体（经验回放 + epsilon-greedy）
✅ RL策略（状态空间：技术指标+持仓+现金）
✅ 动作空间：持有/买入/卖出
✅ 奖励函数：基于投资组合价值变化
✅ 模型保存/加载
✅ 单元测试：12个测试用例全部通过

项目结构

aitrade/
├── aitrade/
│   ├── __init__.py          # 包初始化
│   ├── config/              # 配置管理
│   │   ├── __init__.py
│   │   └── settings.py      # Pydantic配置
│   ├── core/                # 核心模块
│   │   ├── __init__.py
│   │   └── container.py     # 依赖注入容器
│   ├── utils/               # 工具模块
│   │   ├── __init__.py
│   │   └── logger.py        # 结构化日志
│   ├── db/                  # 数据库
│   │   ├── __init__.py
│   │   ├── session.py       # 数据库会话
│   │   └── models.py        # 数据模型
│   ├── data/                # 数据采集
│   │   ├── __init__.py
│   │   ├── base.py          # 数据源基类
│   │   ├── tushare_source.py # Tushare数据源
│   │   ├── akshare_source.py # AKShare数据源
│   │   ├── cleaner.py       # 数据清洗
│   │   └── storage.py       # 数据存储
│   ├── strategy/            # 策略引擎
│   │   ├── __init__.py
│   │   ├── base.py          # 策略基类
│   │   ├── macd.py          # MACD策略
│   │   └── loader.py        # 策略加载器
│   ├── backtest/            # 回测引擎
│   │   ├── __init__.py
│   │   └── engine.py        # 回测引擎
│   ├── risk/                # 风控引擎
│   │   ├── __init__.py
│   │   └── rules.py         # 风控规则
│   ├── execution/           # 交易执行
│   │   ├── __init__.py
│   │   └── broker.py        # 券商模块
│   └── analysis/            # 分析模块
├── alembic/                 # 数据库迁移
├── tests/                   # 测试
├── doc/                     # 文档
├── pyproject.toml           # 项目配置
├── docker-compose.yml       # Docker编排
├── Dockerfile               # Docker镜像
└── .github/workflows/       # CI/CD

快速开始

1. 环境准备

# 克隆项目
git clone <repo-url>
cd aitrade

# 创建虚拟环境
python -m venv .venv
source .venv/bin/activate

# 安装依赖
pip install -e ".[dev]"

2. 配置环境变量

cp .env.example .env
# 编辑.env文件，填入实际配置

3. 启动服务

# 使用Docker Compose启动所有服务
docker-compose up -d

# 或单独启动服务
docker-compose up -d mysql redis

4. 数据库迁移

# 初始化数据库
alembic upgrade head

使用示例

数据采集

from aitrade.data import TushareSource, DataCleaner, DataStorage

# 创建数据源
source = TushareSource(token="your_tushare_token")

# 获取日线数据
df = source.get_daily("600519.SH", start_date="20230101", end_date="20231231")

# 清洗数据
cleaner = DataCleaner()
df = cleaner.clean_daily(df)

# 存储数据
storage = DataStorage()
storage.save_daily(df)

策略回测

from aitrade.data import TushareSource
from aitrade.strategy import MACDStrategy
from aitrade.backtest import BacktestEngine

# 获取数据
source = TushareSource(token="your_tushare_token")
data = source.get_daily("600519.SH", start_date="20230101", end_date="20231231")

# 创建策略
strategy = MACDStrategy(fast_period=12, slow_period=26, signal_period=9)

# 运行回测
engine = BacktestEngine(initial_cash=1000000, commission=0.0002, slippage=0.01)
result = engine.run(strategy, data)

# 查看结果
print(f"总收益率: {result.total_return:.2%}")
print(f"年化收益率: {result.annual_return:.2%}")
print(f"夏普比率: {result.sharpe_ratio:.2f}")
print(f"最大回撤: {result.max_drawdown:.2%}")
print(f"胜率: {result.win_rate:.2%}")

风控管理

from aitrade.risk.rules import RiskManager, MaxDrawdownRule, MaxPositionRule

# 创建风控管理器
manager = RiskManager()

# 添加规则
manager.add_rule(MaxDrawdownRule(threshold=0.1))
manager.add_rule(MaxPositionRule(max_ratio=0.3))

# 检查风控
context = {
    "current_drawdown": 0.05,
    "position_value": 200000,
    "total_value": 1000000,
}
reason = manager.check(context)
if reason:
    print(f"风控触发: {reason}")

开发

运行测试

pytest

代码检查

ruff check aitrade/

代码格式化

black aitrade/

文档

详细设计文档请参考 doc/ 目录。

License

MIT