# ai_quantification **Repository Path**: chobit/ai_quantification ## Basic Information - **Project Name**: ai_quantification - **Description**: python ai量化 - **Primary Language**: Unknown - **License**: Not specified - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2026-06-23 - **Last Updated**: 2026-06-23 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # 币安合约量化交易系统 一个基于Python的币安合约量化交易系统框架，提供完整的底层基础设施和清晰的策略接口。 ## 功能特性 ### 核心功能 - **实时行情获取**：通过币安API获取实时行情数据 - **WebSocket推送**：实时K线、Tick、深度数据推送 - **合约账户管理**：账户余额、持仓信息、未实现盈亏 - **下单与仓位管理**：市价单、限价单、止损止盈单 - **异常重连机制**：WebSocket断线自动重连 ### 策略系统 - **策略基类**：提供完整的策略接口和回调钩子 - **策略管理**：策略的加载、启动、停止、监控 - **参数管理**：策略参数的配置和管理 - **变量管理**：策略变量的持久化和恢复 ### 风险控制 - **仓位控制**：最大持仓限制、单笔限额 - **资金管理**：每日亏损限额、最大回撤控制 - **风险监控**：实时风险指标监控和预警 - **自动止损**：触发风控规则时自动平仓 ### 回测系统 - **历史数据回测**：支持多周期、多交易对回测 - **历史数据下载**：从币安API下载历史K线数据，支持本地缓存 - **绩效评估**：收益率、夏普比率、最大回撤等指标 - **交易记录**：详细的交易记录和绩效分析 - **可视化分析**：权益曲线、回撤曲线等图表 ### 监控系统 - **日志记录**：多级别、多通道日志记录 - **性能监控**：系统性能指标监控 - **可视化面板**：实时监控和状态可视化 - **通知系统**：风险预警和交易信号通知 ## 系统架构 ``` binance_bot/ ├── config/ # 配置文件 ├── core/ # 核心模块 │ ├── engine.py # 主引擎 │ ├── event.py # 事件处理 │ └── config.py # 配置管理 ├── gateway/ # 接口网关 │ ├── binance_gateway.py # 币安接口 │ └── websocket.py # WebSocket管理 ├── strategy/ # 策略模块 │ ├── base_strategy.py # 策略基类 │ └── template/ # 策略模板 ├── backtest/ # 回测模块 │ ├── engine.py # 回测引擎 │ ├── data_manager.py # 数据管理 │ └── data_downloader.py # 历史数据下载 ├── risk/ # 风控模块 │ ├── risk_manager.py # 风险管理 │ └── position_manager.py # 仓位管理 ├── monitor/ # 监控模块 │ ├── dashboard.py # 可视化面板 │ └── logger.py # 日志系统 ├── utils/ # 工具函数 ├── tests/ # 单元测试 ├── data/ # 数据存储 └── logs/ # 日志文件 ``` ## 快速开始 ### 1. 安装依赖 ```bash pip install -r requirements.txt ``` ### 2. 配置系统 编辑 `config/config.yaml` 文件，配置币安API密钥和其他参数： ```yaml binance: api_key: "your_api_key_here" api_secret: "your_api_secret_here" testnet: true # 使用测试网 trading: symbols: ["BTCUSDT", "ETHUSDT"] default_leverage: 10 risk: max_position_size: 1000000 max_daily_loss: 50000 max_drawdown: 0.1 ``` ### 3. 运行实盘交易 ```bash python main.py --mode live ``` ### 4. 运行回测 ```bash python main.py --mode backtest --strategy SimpleMA ``` ## 编写策略 ### 策略基类 所有策略必须继承 `BaseStrategy` 类，并实现以下方法： ```python from strategy.base_strategy import BaseStrategy from utils.data_types import * class MyStrategy(BaseStrategy): name = "MyStrategy" def on_init(self): """策略初始化""" pass def on_start(self): """策略启动""" pass def on_stop(self): """策略停止""" pass def on_tick(self, tick: TickData): """Tick数据更新""" pass def on_bar(self, bar: BarData): """K线数据更新""" pass def on_order(self, order: OrderData): """订单状态更新""" pass def on_trade(self, trade: TradeData): """成交信息更新""" pass def on_position(self, position: PositionData): """持仓信息更新""" pass def on_account(self, account: AccountData): """账户信息更新""" pass ``` ### 交易接口 ```python # 买入开仓 order_id = self.buy("BTCUSDT", 50000, 0.1, OrderType.LIMIT) # 卖出平仓 order_id = self.sell("BTCUSDT", 51000, 0.1, OrderType.LIMIT) # 卖出开仓（做空） order_id = self.short("BTCUSDT", 51000, 0.1, OrderType.LIMIT) # 买入平仓（平空） order_id = self.cover("BTCUSDT", 50000, 0.1, OrderType.LIMIT) # 取消订单 self.cancel_order(order_id) # 获取持仓 position = self.get_position("BTCUSDT") # 获取账户信息 account = self.get_account() ``` ### 数据接口 ```python # 获取Tick数据 tick = self.get_tick("BTCUSDT") # 获取K线数据 bars = self.get_bars("BTCUSDT", 100) # 获取最近100根K线 # 获取合约信息 contract = self.get_contract("BTCUSDT") ``` ## 风险控制 ### 仓位控制 ```python # 设置杠杆 self.set_leverage("BTCUSDT", 10) # 设置保证金模式 self.set_margin_type("BTCUSDT", "CROSSED") ``` ### 止损止盈 ```python # 止损止盈在策略中实现 def _check_stop_loss_profit(self): position = self.get_position("BTCUSDT") if not position: return tick = self.get_tick("BTCUSDT") current_price = tick.last_price entry_price = position.price # 计算盈亏比例 if position.direction == PositionSide.LONG: pnl_ratio = (current_price - entry_price) / entry_price else: pnl_ratio = (entry_price - current_price) / entry_price # 止损检查 if pnl_ratio < -0.02: # 亏损2% if position.direction == PositionSide.LONG: self.sell("BTCUSDT", 0, position.volume, OrderType.MARKET) else: self.cover("BTCUSDT", 0, position.volume, OrderType.MARKET) # 止盈检查 elif pnl_ratio > 0.05: # 盈利5% if position.direction == PositionSide.LONG: self.sell("BTCUSDT", 0, position.volume, OrderType.MARKET) else: self.cover("BTCUSDT", 0, position.volume, OrderType.MARKET) ``` ## 回测系统 ### 历史数据下载 ```python from backtest.data_downloader import DataDownloader from datetime import datetime # 创建下载器 downloader = DataDownloader(config) # 下载数据 data = await downloader.download_data( symbols=["BTCUSDT", "ETHUSDT"], intervals=["1m", "5m"], start_date=datetime(2024, 1, 1), end_date=datetime(2024, 12, 31), save_to_csv=True ) # 使用缓存下载（自动跳过已下载的数据） data = await downloader.download_and_cache( symbols=["BTCUSDT"], intervals=["1h"], start_date=datetime(2024, 1, 1), end_date=datetime(2024, 12, 31), force_download=False ) # 从CSV加载数据 df = downloader.load_from_csv("BTCUSDT", "1m") # 获取可用数据 available = downloader.get_available_data() print(f"可用数据: {available}") ``` ### 运行回测 ```python from backtest.engine import BacktestEngine from strategy.template.simple_strategy import SimpleStrategy # 创建回测引擎 engine = BacktestEngine(config) # 设置策略 engine.set_strategy(SimpleStrategy, { "fast_period": 10, "slow_period": 30, "trade_volume": 0.1, "symbol": "BTCUSDT", "interval": "1m" }) # 加载数据 await engine.load_data(["BTCUSDT"], ["1m"]) # 运行回测 engine.run_backtest() # 获取结果 performance = engine.get_performance_stats() print(f"总收益率: {performance['total_return']:.2%}") print(f"最大回撤: {performance['max_drawdown']:.2%}") print(f"夏普比率: {performance['sharpe_ratio']:.2f}") ``` ## 监控系统 ### 日志记录 ```python from monitor.logger import get_logger logger = get_logger("MyStrategy") logger.info("策略启动") logger.error("发生错误") ``` ### 可视化面板 ```python from monitor.dashboard import Dashboard dashboard = Dashboard(config) await dashboard.start() # 更新数据 dashboard.update_tick(tick) dashboard.update_position(position) dashboard.update_account(account) ``` ## 配置说明 ### 币安API配置 ```yaml binance: api_key: "your_api_key_here" api_secret: "your_api_secret_here" testnet: true # 使用测试网 base_url: "https://testnet.binancefuture.com" ws_url: "wss://testnet.binancefuture.com/ws" ``` ### 交易配置 ```yaml trading: symbols: ["BTCUSDT", "ETHUSDT"] default_leverage: 10 max_leverage: 20 margin_mode: "CROSSED" # CROSSED/ISOLATED position_mode: "ONE_WAY" # ONE_WAY/HEDGE ``` ### 风控配置 ```yaml risk: max_position_size: 1000000 max_position_per_symbol: 100000 max_daily_loss: 50000 max_drawdown: 0.1 max_order_value: 10000 min_order_value: 10 default_stop_loss: 0.02 ``` ### 回测配置 ```yaml backtest: data_source: "binance" data_path: "./data" start_date: "2024-01-01" end_date: "2024-12-31" initial_capital: 100000 maker_fee: 0.0002 taker_fee: 0.0004 slippage: 0.001 ``` ## 注意事项 1. **API安全**：请妥善保管API密钥，不要泄露给他人 2. **测试网**：开发测试时请使用测试网，避免真实资金损失 3. **风险控制**：务必设置合理的风控参数，控制交易风险 4. **系统监控**：定期检查系统日志和性能指标 5. **数据备份**：定期备份交易数据和配置文件 ## 开发计划 - [x] 核心框架开发 - [x] 币安API集成 - [x] 策略基类开发 - [x] 风险控制模块 - [x] 回测系统开发 - [x] 历史数据下载器 - [x] 日志系统开发 - [ ] 可视化面板完善 - [ ] 更多策略模板 - [ ] 性能优化 - [ ] 文档完善 ## 许可证 MIT License ## 联系方式 如有问题或建议，请提交Issue或Pull Request。