本文档面向开发者,涵盖项目架构、开发环境搭建和贡献指引。
- Python >= 3.12
- uv (推荐) 或 pip
git clone https://github.com/Aixtrade/QuantTrader.git
cd XQTrader
# 使用 uv 安装依赖
uv sync --group dev
# 或使用 pip
pip install -e ".[dev]"xqtrader/
├── strategies/ # 策略系统 - BaseStrategy 基类和动态加载器
├── engine/ # 执行引擎 - BacktestEngine / RealtimeEngine
├── accounts/ # 账户管理 - SimulatedAccount / FuturesSimulatedAccount
├── traders/ # 交易器 - EventsTrader / FuturesTrader
├── data/ # 数据服务 - DataCenterService + CCXT 适配器
│ └── adapters/ # 交易所适配器 - CCXTAdapter / BinanceAdapter
├── indicators/ # 技术指标 - 基于 talipp 的 60+ 指标
├── risk/ # 风控系统 - RiskManager (WARNING/CRITICAL 分级)
├── reports/ # 报告生成 - BacktestReport / TradeRecord
└── config/ # 配置管理
市场数据 (OHLCV) → 指标引擎 → 策略上下文 → 策略执行
↓ ↓
DataCenterService StrategyResult
↓
账户更新 ← 交易执行 ← 风控检查 ← 交易信号
↓
回测报告
StrategySignal.action:BUY/SELL/HOLD(通用) 或LONG/SHORT/CLOSE_LONG/CLOSE_SHORT/CLOSE(永续合约)PositionSide:LONG/SHORT(双向持仓)ExecutionMode:BACKTEST/PAPER/LIVERiskLevel:NORMAL/WARNING/CRITICAL
继承 BaseStrategy 并实现 execute(context: StrategyContext) -> StrategyResult:
from xqtrader.strategies.base import (
BaseStrategy,
StrategyContext,
StrategyResult,
)
class MyStrategy(BaseStrategy):
def __init__(self):
super().__init__(name="my_strategy", version="1.0.0")
def execute(self, context: StrategyContext) -> StrategyResult:
close_prices = context.market_data.get("close", [])
if not close_prices:
return StrategyResult(
signals=[self.create_signal("HOLD", context.symbol)],
indicators={},
metadata={},
execution_time=0.0,
success=True,
)
signal = self.create_signal("LONG", context.symbol, confidence=0.8)
return StrategyResult(
signals=[signal],
indicators={},
metadata={},
execution_time=0.0,
success=True,
)| 合约类型 | 信号 |
|---|---|
| 永续/期货 | LONG, SHORT, CLOSE_LONG, CLOSE_SHORT, CLOSE |
| 事件合约 | UP, DOWN, HOLD |
事件合约交易器兼容
LONG/SHORT/BUY/SELL并自动映射到UP/DOWN。
- 使用
FuturesSimulatedAccount+HedgePositionManager(双向持仓) - 保证金通过
lock_margin/release_margin管理 - 盈亏计算: 多仓
(exit - entry) * size,空仓(entry - exit) * size
基于 CCXT 实现,支持 100+ 交易所:
from xqtrader.data import DataCenterService, MarketDataRequest, MarketType
# 现货数据
async with DataCenterService(market_type=MarketType.SPOT) as dc:
data = await dc.get_market_data(MarketDataRequest(symbol="BTC/USDT", interval="1h"))
ticker = await dc.get_ticker("BTC/USDT")
# 永续合约
async with DataCenterService(market_type=MarketType.FUTURES) as dc:
data = await dc.get_market_data(MarketDataRequest(symbol="BTC/USDT", interval="1h"))特性:LRU 缓存 + 熔断器保护。Binance 合约额外支持 fetch_mark_price_klines() 和 fetch_current_funding_rate()。
| 规则 | 警告 | 强平 |
|---|---|---|
| 日亏损 | 3.5% | 5% |
| 最大回撤 | 10% | 15% |
# 运行所有测试
uv run pytest
# 运行特定测试文件
uv run pytest tests/test_data_center.py -v
# 运行特定测试类
uv run pytest tests/test_data_center.py::TestLRUCache -v| 包 | 用途 |
|---|---|
| ccxt | 交易所连接 |
| talipp | 技术指标 |
| numpy | 数值计算 |
| pydantic | 数据验证 |
| httpx | 异步 HTTP 客户端 |
- MVP v0: 单标的回测闭环
- MVP v1: 实时纸交易
- MVP v2: 资金费率优化
- MVP v3: 多资产组合
- MVP v4: 实盘接口
欢迎贡献代码!请阅读 AGENTS.md 了解编码规范。