# miniQMT **Repository Path**: vx-qa/miniQMT ## Basic Information - **Project Name**: miniQMT - **Description**: miniQMT 是一个基于迅投QMT API的量化交易系统,实现了自动化交易策略执行、持仓管理、止盈止损和网格交易等功能。系统支持模拟交易和实盘交易两种模式。 - **Primary Language**: Python - **License**: Not specified - **Default Branch**: main - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 3 - **Forks**: 0 - **Created**: 2025-12-14 - **Last Updated**: 2026-01-25 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # miniQMT - 无人值守量化交易系统
[![Python Version](https://img.shields.io/badge/python-3.8%2B-blue)](https://www.python.org/downloads/) [![License](https://img.shields.io/badge/license-MIT-green)](LICENSE) [![QMT API](https://img.shields.io/badge/QMT-API-orange)](https://dict.thinktrader.net) **基于迅投QMT API的智能量化交易系统** [功能特性](#功能特性) • [快速开始](#快速开始) • [系统架构](#系统架构) • [文档](#文档) • [常见问题](#常见问题)
--- ## 📖 项目简介 miniQMT 是一个专为A股市场设计的**无人值守量化交易系统**,集成了自动化交易策略执行、智能持仓管理、动态止盈止损等功能。系统采用双层存储架构和线程自愈机制,可7x24小时稳定运行。 ### 🎯 核心优势 - **🛡️ 无人值守**: 线程健康监控与自动重启,系统崩溃自动恢复 - **🔄 双层存储**: 内存数据库 + SQLite持久化,高性能与数据安全兼顾 - **🎯 信号分离**: 信号检测与执行分离,策略逻辑清晰可控 - **📈 智能止盈止损**: 动态止盈策略,最大化收益同时控制风险 - **🌐 Web监控**: 实时监控界面,随时掌握系统运行状态 - **🧪 模拟交易**: 完整的模拟交易功能,策略验证零风险 --- ## ✨ 功能特性 ### 交易功能 - ✅ **自动交易执行**: 支持模拟交易和实盘交易 - ✅ **动态止盈止损**: 首次止盈 + 动态止盈双重保护 - ✅ **网格交易**: 可选的网格交易策略 - ✅ **信号验证机制**: 防止重复执行,确保交易安全 ### 系统功能 - ✅ **线程健康监控**: 自动检测线程崩溃并重启 - ✅ **优雅关闭**: 有序关闭各模块,避免数据丢失 - ✅ **超时保护**: API调用超时保护,防止线程阻塞 - ✅ **非交易时段优化**: 自动降低CPU占用(30% → <2%) ### 监控功能 - ✅ **Web实时监控**: 账户信息、持仓列表实时更新 - ✅ **SSE推送**: Server-Sent Events实时数据推送 - ✅ **系统诊断**: 内置多个诊断工具,快速定位问题 --- ## 🚀 快速开始 ### 环境要求 - **Python**: 3.8+ (推荐 3.9) - **操作系统**: Windows (QMT仅支持Windows) - **QMT客户端**: 实盘交易需要安装并登录QMT ### 安装步骤 1. **克隆项目** ```bash git clone https://github.com/your-repo/miniQMT.git cd miniQMT ``` ### 模块化设计原则 系统采用严格的模块化设计,每个py文件对应一个独立功能模块: ``` miniQMT/ ├── config.py # 集中管理所有可配置参数 ├── logger.py # 统一日志管理(自动清理) ├── main.py # 系统启动入口和线程管理 ├── data_manager.py # 历史数据获取与存储(xtdata接口) ├── indicator_calculator.py # 技术指标计算(MACD、均线等) ├── position_manager.py # 持仓管理核心(内存+SQLite双层架构) ├── trading_executor.py # 交易执行器(xttrader接口) ├── strategy.py # 交易策略逻辑 ├── web_server.py # RESTful API服务(Flask) ├── easy_qmt_trader.py # QMT交易API封装 ├── utils.py / Methods.py # 工具函数集合 ├── MyTT.py # 技术指标计算工具 ├── account_config.json # 账户配置文件(敏感信息) ├── stock_pool.json # 股票池配置 ├── data/ # 数据存储目录 │ ├── trading.db # SQLite数据库 │ └── stock2buy.json # 备选股票池 ├── logs/ # 日志文件目录 ├── test/ # 测试文件目录 │ ├── unit_test.py # 单元测试 │ ├── StopLossTakeProfitTest.py # 止盈止损测试 │ ├── GridTradingTest.py # 网格交易测试 │ ├── test_qmt_position_interface.py # QMT持仓接口测试 │ ├── test_qmt_path_config.py # QMT路径配置测试 │ └── test_qmt_fix_verification.py # QMT修复验证测试 ├── web1.0/ # Web前端界面 │ ├── index.html # 主页面 │ ├── script.js # 前端逻辑 │ └── readme.md # 前端说明 ├── xtquant/ # 迅投QMT SDK └── CLAUDE.md # 本文档 ``` ### 模块依赖关系 ```mermaid graph LR config[config.py] --> |配置参数| all[所有模块] logger[logger.py] --> |日志记录| all main[main.py] --> |初始化| dm[data_manager] main --> |初始化| pm[position_manager] main --> |初始化| st[strategy] main --> |初始化| ws[web_server] dm --> |数据| ic[indicator_calculator] ic --> |指标| st pm --> |持仓信息| st pm --> |持仓信息| ws pm --> |交易API| eqt[easy_qmt_trader] st --> |交易指令| te[trading_executor] te --> |执行交易| eqt ws --> |API调用| st ws --> |API调用| te ws --> |API调用| pm style config fill:#ffd,stroke:#333 style logger fill:#ffd,stroke:#333 style main fill:#f9f,stroke:#333,stroke-width:2px ``` ### 线程架构 系统采用多线程并发架构,各线程职责如下: | 线程名称 | 启动位置 | 主要职责 | 运行频率 | |---------|---------|---------|---------| | **数据更新线程** | `data_manager.start_data_update_thread()` | 更新股票池中所有股票的最新行情数据 | 每60秒 | | **持仓监控线程** | `position_manager.start_position_monitor_thread()` | 1. 同步实盘持仓到内存
2. 更新当前价格
3. 检测止盈止损信号 | 每3秒 | | **策略执行线程** | `strategy.start_strategy_thread()` | 1. 获取待处理信号
2. 执行交易决策
3. 网格交易检查 | 每5秒 | | **定时同步线程** | `position_manager.start_sync_thread()` | 将内存数据库的关键字段同步到SQLite | 每5秒 | | **日志清理线程** | `logger.schedule_log_cleanup()` | 清理过期日志文件 | 每天凌晨 | | **Web服务线程** | `web_server.start_web_server()` | 提供RESTful API和前端页面服务 | 持续运行 | **线程安全机制**: - 使用 `threading.Lock()` 保护共享数据结构 - SQLite 连接使用 `check_same_thread=False` - 内存数据库操作使用版本号机制避免竞态 ### 数据流架构 **实盘模式数据流**: ```mermaid sequenceDiagram participant QMT as QMT实盘账户 participant PM as position_manager participant MEM as 内存数据库 participant DB as SQLite数据库 loop 每3秒 QMT->>PM: qmt_trader.position() PM->>MEM: _sync_real_positions_to_memory() Note over MEM: volume, available, cost_price end loop 每5秒 MEM->>DB: _sync_memory_to_db() Note over DB: 持久化: open_date, profit_triggered, etc. end loop 实时更新 PM->>MEM: update_all_positions_price() Note over MEM: current_price, market_value, profit_ratio end ``` **模拟模式数据流**: ```mermaid sequenceDiagram participant UI as Web界面 participant TE as trading_executor participant PM as position_manager participant MEM as 内存数据库 UI->>TE: 模拟买入/卖出指令 TE->>PM: simulate_buy/sell_position() PM->>MEM: 直接修改持仓数据 Note over MEM: 跳过QMT接口
资金从SIMULATION_BALANCE扣除/增加 PM->>MEM: 生成模拟订单ID: SIM{timestamp}{counter} ``` ### 信号检测与执行流程 系统采用**信号检测与执行分离**的设计模式: ```mermaid flowchart TD A[持仓监控线程] --> B{检测交易信号} B --> C[止损信号] B --> D[首次止盈信号] B --> E[动态止盈信号] C --> F[写入 latest_signals 队列] D --> F E --> F F --> G[策略执行线程] G --> H{ENABLE_AUTO_TRADING?} H -->|False| I[仅监控不执行] H -->|True| J[get_pending_signals] J --> K{validate_trading_signal} K -->|验证失败| L[拒绝执行] K -->|验证通过| M{模拟/实盘模式?} M -->|模拟| N[simulate_sell_position] M -->|实盘| O[trading_executor.sell_stock] O --> P[QMT API执行] N --> Q[直接修改内存数据库] style H fill:#faa,stroke:#333 style M fill:#afa,stroke:#333 ``` **关键设计要点**: 1. 监控线程**始终运行**,持续检测信号(即使 ENABLE_AUTO_TRADING=False) 2. `ENABLE_AUTO_TRADING` 开关只控制**是否执行**检测到的信号 3. 每个信号都要经过 `validate_trading_signal()` 验证,防止重复执行 4. 信号验证包括:持仓数据一致性、信号是否过期、是否已处理 ### 关键设计模式 1. **单例模式**: 所有核心模块通过 `get_*()` 函数返回单例实例 ```python # 示例 from position_manager import PositionManager def get_position_manager(): global _position_manager_instance if _position_manager_instance is None: _position_manager_instance = PositionManager() return _position_manager_instance ``` 2. **双层存储架构**: - 内存数据库(`sqlite3.connect(":memory:")`)存储高频更新数据 - SQLite文件数据库持久化关键状态(避免重启丢失) 3. **信号检测与执行分离**: - 监控线程负责信号检测(始终运行) - `ENABLE_AUTO_TRADING`开关控制是否执行交易 2. **安装依赖** ```bash pip install pandas numpy flask flask-cors xtquant mootdx TA-Lib ``` **关键依赖说明**: - `pandas`, `numpy`: 数据处理和计算 - `flask`, `flask-cors`: Web服务框架 - `xtquant`: 迅投QMT交易和行情接口(需要安装QMT客户端) - `mootdx`: 备用行情数据源 - `talib` (可选): 高级技术指标计算 3. **配置账户** (创建 `account_config.json`) ```json { "account_id": "您的交易账号", "account_type": "STOCK", "qmt_path": "C:/光大证券金阳光QMT实盘/userdata_mini" } ``` 4. **配置股票池** (可选,创建 `stock_pool.json`) ```json [ "000001.SZ", "600036.SH", "000333.SZ" ] ``` 5. **启动系统** ```bash python main.py ``` 6. **访问Web界面** ``` http://localhost:5000 ``` ### ⚠️ 首次运行建议 **强烈建议先使用模拟模式测试**: 1. 确认 `config.py` 中 `ENABLE_SIMULATION_MODE = True` 2. 运行无人值守功能测试: `python test/test_unattended_operation.py` 3. 观察系统运行稳定后,再切换到实盘模式 --- ## 🏗️ 系统架构 ### 核心设计 ``` ┌─────────────────────────────────────────────────────────┐ │ 线程监控器 (60秒) │ │ 检测线程崩溃 → 自动重启 │ └─────────────────────────────────────────────────────────┘ ↓ ┌──────────────┬──────────────┬──────────────┬────────────┐ │ 数据更新线程 │ 持仓监控线程 │ 策略执行线程 │ Web服务 │ │ (60秒) │ (3秒) │ (5秒) │ (持续) │ └──────────────┴──────────────┴──────────────┴────────────┘ ↓ ↓ ↓ ↓ ┌─────────────────────────────────────────────────────────┐ │ 双层存储架构 │ │ 内存数据库 (高频更新) ←→ SQLite (持久化) │ └─────────────────────────────────────────────────────────┘ ``` ### 数据流 **实盘模式**: ``` QMT实盘账户 → 持仓同步(10秒) → 内存数据库 → 定时同步(15秒) → SQLite ``` **模拟模式**: ``` Web界面 → 交易执行器 → 内存数据库 (跳过QMT接口) ``` 详细架构说明请查看 [ARCHITECTURE.md](ARCHITECTURE.md) --- ## 📊 核心配置 ### 功能开关 (config.py) ```python # 交易模式 ENABLE_SIMULATION_MODE = True # True=模拟, False=实盘 ⚠️ ENABLE_AUTO_TRADING = False # 自动交易执行开关 ⚠️ # 策略功能 ENABLE_DYNAMIC_STOP_PROFIT = True # 止盈止损功能 ENABLE_GRID_TRADING = False # 网格交易功能 # 系统功能 ENABLE_THREAD_MONITOR = True # 线程健康监控(无人值守必需)⭐ DEBUG = False # 调试模式 ``` ### 常用命令速查 ```bash # 启动系统 python main.py # 运行测试 python -m pytest test/unit_test.py # 单元测试 python test/StopLossTakeProfitTest.py # 止盈止损测试 python test/GridTradingTest.py # 网格交易测试 python test/easy_qmt_trader_test.py # QMT接口测试 python test/test_qmt_position_interface.py # 持仓接口测试 python test/test_qmt_path_config.py # 路径配置测试 # 清理日志 python -c "from logger import clean_old_logs; clean_old_logs()" # 检查数据库 sqlite3 data/trading.db ".tables" # 查看所有表 sqlite3 data/trading.db "SELECT * FROM positions;" # 查看持仓 ``` ### Windows 环境特别注意 - 使用 Anaconda Python 3.10 虚拟环境(默认) - 日志文件使用 UTF-8 编码(避免中文乱码) - 路径使用正斜杠 `/` 或双反斜杠 `\\` ### 止盈止损配置 ```python # 止损 STOP_LOSS_RATIO = -0.075 # 成本价下跌7.5%触发止损 # 首次止盈 INITIAL_TAKE_PROFIT_RATIO = 0.06 # 盈利6%触发 INITIAL_TAKE_PROFIT_RATIO_PERCENTAGE = 0.6 # 卖出60% # 动态止盈 DYNAMIC_TAKE_PROFIT = [ (0.05, 0.96), # 最高浮盈5%时,止盈位=最高价*96% (0.10, 0.93), # 最高浮盈10%时,止盈位=最高价*93% (0.15, 0.90), (0.20, 0.87), (0.30, 0.85), ] ``` --- ## 📚 文档 - **[CLAUDE.md](CLAUDE.md)** - 开发指南(面向AI助手和开发者) - **[ARCHITECTURE.md](ARCHITECTURE.md)** - 详细架构说明 - **[docs/quick_start_unattended.md](docs/quick_start_unattended.md)** - 无人值守运行指南 - **[docs/unattended_operation_summary.md](docs/unattended_operation_summary.md)** - 线程监控详解 --- ## 🧪 测试 ### 推荐测试顺序 ```bash # 1. 无人值守功能测试(验证线程自愈机制) python test/test_unattended_operation.py # 2. 系统综合测试 python test/comprehensive_test.py # 3. 止盈止损测试 python test/test_stop_loss_buy_param.py # 4. Web数据刷新测试 python test/test_web_data_refresh.py # QMT交易接口测试 python test/easy_qmt_trader_test.py # QMT路径配置测试(验证路径自动检测) python test/test_qmt_path_config.py # QMT修复验证测试(验证接口修复) python test/test_qmt_fix_verification.py ``` ### 系统诊断工具 ```bash # 检查系统状态 python test/check_system_status.py # 诊断QMT连接 python test/diagnose_qmt_connection.py # 诊断系统问题 python test/diagnose_system_issues.py ``` ### 调试技巧 **启用详细日志**: ```python # config.py DEBUG = True LOG_LEVEL = "DEBUG" ``` **测试模拟交易**: ```python # config.py ENABLE_SIMULATION_MODE = True DEBUG_SIMU_STOCK_DATA = True # 绕过交易时间限制 ``` **监控关键数据流**: ```python # 查看内存持仓数据 position_manager.get_all_positions() # 查看待执行信号 position_manager.get_pending_signals() # 检查账户信息 position_manager.get_account_info() # 查看信号队列 position_manager.latest_signals ``` **调试止盈止损逻辑**: ```python # 在 strategy.py 中添加断点 def execute_trading_signal_direct(self, stock_code, signal_type, signal_info): # 在此处设置断点,观察信号内容 logger.debug(f"执行信号: {signal_type}, 信息: {signal_info}") ``` ## API接口说明 ### Web API端点 (web_server.py) **系统状态**: - `GET /api/status` - 获取系统运行状态 - 返回:账户信息、监控状态、全局开关状态 - `GET /api/connection/status` - 检查QMT连接状态 **持仓管理**: - `GET /api/positions` - 获取所有持仓 - 返回:持仓列表、数据版本号 - `GET /api/positions/` - 获取单只股票持仓 - `GET /api/positions/stream` - SSE实时推送持仓数据 **交易操作**: - `POST /api/actions/execute_buy` - 执行买入操作 - 参数:`stock_code`, `amount`, `strategy`(可选) - `POST /api/actions/execute_sell` - 执行卖出操作 - 参数:`stock_code`, `volume`, `strategy`(可选) - `POST /api/actions/execute_trading_signal` - 执行指定交易信号 - 参数:`stock_code`, `signal_type`, `signal_info` **配置管理**: - `GET /api/config` - 获取系统配置 - `POST /api/config/update` - 更新配置参数 - 支持动态更新:止盈止损比例、仓位限制等 **信号查询**: - `GET /api/signals/pending` - 获取待处理信号列表 - `GET /api/signals/latest/` - 获取指定股票的最新信号 **数据查询**: - `GET /api/data/history/` - 获取历史K线数据 - `GET /api/data/realtime/` - 获取实时行情 ## QMT API集成要点 ### xtdata(行情接口) ```python # data_manager.py import xtquant.xtdata as xt # 连接行情服务 xt.connect() # 获取历史数据 xt.get_market_data( field_list=['open', 'high', 'low', 'close', 'volume'], stock_list=['000001.SZ'], period='1d', start_time='20230101', end_time='20231231' ) # 获取实时Tick数据 xt.get_full_tick(['000001.SZ']) ``` ### xttrader(交易接口) ```python # easy_qmt_trader.py from xtquant.xttrader import XtQuantTrader from xtquant.xttype import StockAccount # 创建交易对象 xt_trader = XtQuantTrader(path, session_id) # 启动交易线程 xt_trader.start() # 连接账户 acc = StockAccount(account_id, account_type) xt_trader.connect() # 下单 xt_trader.order_stock( acc, stock_code, order_type, # 23=限价买入, 24=限价卖出 order_volume, order_price ) # 查询持仓 xt_trader.query_stock_positions(acc) # 查询资产 xt_trader.query_stock_asset(acc) ``` ## 数据库表结构 ### positions (持仓表 - 内存+SQLite) ```sql CREATE TABLE positions ( stock_code TEXT PRIMARY KEY, -- 股票代码 stock_name TEXT, -- 股票名称 volume REAL, -- 持仓数量(来自QMT) available REAL, -- 可用数量(来自QMT) cost_price REAL, -- 成本价(来自QMT) base_cost_price REAL, -- 基础成本价 current_price REAL, -- 当前价(来自行情接口) market_value REAL, -- 市值(计算得出) profit_ratio REAL, -- 盈亏比例(计算得出) last_update TIMESTAMP, -- 最后更新时间 open_date TIMESTAMP, -- 开仓日期(持久化) profit_triggered BOOLEAN, -- 是否触发首次止盈(持久化) highest_price REAL, -- 历史最高价(持久化) stop_loss_price REAL, -- 止损价(持久化) profit_breakout_triggered BOOLEAN, -- 是否触发突破止盈 breakout_highest_price REAL -- 突破后最高价 ) ``` **数据来源说明**: | 字段 | 数据来源 | 更新时机 | |------|---------|---------| | `stock_code`, `volume`, `available`, `cost_price` | QMT实盘 `qmt_trader.position()` | 每3秒同步一次 | | `current_price` | `data_manager.get_latest_data()` | 实时更新 | | `market_value`, `profit_ratio` | 计算 | 价格更新时重新计算 | | `open_date`, `profit_triggered`, `highest_price`, `stop_loss_price` | 持久化字段 | 策略触发时更新并立即同步到SQLite | ### trade_records (交易记录表) ```sql CREATE TABLE trade_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, stock_code TEXT, -- 股票代码 stock_name TEXT, -- 股票名称 trade_time TIMESTAMP, -- 交易时间 trade_type TEXT, -- BUY, SELL price REAL, -- 交易价格 volume INTEGER, -- 交易数量 amount REAL, -- 交易金额 trade_id TEXT, -- 订单ID commission REAL, -- 手续费 strategy TEXT -- 策略标识(simu/auto_partial/stop_loss/grid等) ) ``` ### stock_daily_data (历史K线数据表) ```sql CREATE TABLE stock_daily_data ( stock_code TEXT, stock_name TEXT, date TEXT, open REAL, high REAL, low REAL, close REAL, volume REAL, amount REAL, PRIMARY KEY (stock_code, date) ) ``` ### stock_indicators (指标数据表) ```sql CREATE TABLE stock_indicators ( stock_code TEXT, date TEXT, ma10 REAL, ma20 REAL, ma30 REAL, ma60 REAL, macd REAL, macd_signal REAL, macd_hist REAL, PRIMARY KEY (stock_code, date) ) ``` --- ## ❓ 常见问题 ### Q1: 如何切换到实盘交易? **A**: 修改 `config.py`: ```python ENABLE_SIMULATION_MODE = False # 切换到实盘 ENABLE_AUTO_TRADING = True # 启用自动交易 ``` **⚠️ 注意**: 确保QMT客户端已启动并登录! ### Q2: 系统崩溃后会自动恢复吗? **A**: 会的! 系统内置线程健康监控,每60秒检查一次线程状态,检测到崩溃立即重启。详见 [无人值守运行文档](docs/quick_start_unattended.md) ### Q3: 如何查看系统运行日志? **A**: 日志文件位于 `logs/qmt_trading.log`,可以使用: ```bash # 实时查看日志 tail -f logs/qmt_trading.log # 查看最近100行 tail -n 100 logs/qmt_trading.log ``` ### Q4: 模拟交易和实盘交易有什么区别? **A**: - **模拟交易**: 不调用QMT API,使用虚拟资金,无交易时间限制 - **实盘交易**: 通过QMT API执行真实订单,使用实际账户资金 ### Q5: 如何添加新的股票到股票池? **A**: 编辑 `stock_pool.json` 文件,添加股票代码(格式: `股票代码.交易所`): ```json [ "000001.SZ", // 深圳交易所 "600036.SH", // 上海交易所 "新增股票.SZ" ] ``` 更多问题请查看 [CLAUDE.md - 常见问题](CLAUDE.md#常见问题与解决方案) --- ## 🛡️ 风险提示 **⚠️ 重要提醒**: 1. **量化交易有风险,投资需谨慎** 2. **本系统仅供学习研究使用,不构成投资建议** 3. **实盘交易前请充分测试策略,确保理解系统运行逻辑** 4. **建议设置合理的止损比例,控制单日最大亏损** 5. **定期检查系统运行状态,关注异常日志** --- ## 📄 许可证 本项目采用 MIT 许可证 - 详见 [LICENSE](LICENSE) 文件 --- ## 🤝 贡献 欢迎提交 Issue 和 Pull Request! --- ## 📧 联系方式 如有问题或建议,请通过以下方式联系: - 提交 [GitHub Issue](https://github.com/your-repo/miniQMT/issues) - 邮箱: your-email@example.com ---
**⭐ 如果这个项目对你有帮助,请给个Star支持一下! ⭐** Made with ❤️ by miniQMT Team