OneQuant 4.0 量化交易系统

版本：4.0.1 | 发布日期：2026-06-10 | 许可：本地部署，个人使用

OneQuant 4.0 是一个本地运行的 Flask 单页应用，面向 A 股个人投资者和量化学习者。它把策略回测、技术分析、模拟交易、风控管理整合在一个 Web 界面里，同时对所有功能暴露 REST API 供程序化调用。

🤖 AI 使用约束（AI Behavior Rules）

禁止事项（违反将导致输出不可信）：

1. ❌ 禁止编造数据：所有行情/回测/K线数据必须通过实际 API 调用获取，绝对不能凭空生成
2. ❌ 禁止声称实时连接：系统通过 REST API 轮询获取数据，非事件驱动/毫秒级推送，不可声称"实时"
3. ❌ 禁止保证投资收益：回测结果仅供参考，不可暗示或保证任何策略能产生盈利
4. ❌ 禁止操作真实券商：默认 mock_broker 是模拟环境，不可绕过用户确认直接操作实盘
5. ❌ 禁止猜测 API 参数：如果用户提供的股票代码/参数不在支持的范围内，必须明确告知不支持

强制规则：

• 每个数据输出必须注明数据来源（AkShare/JQData/RQData）和获取时间
• 回测结果必须附带"过去表现不代表未来收益"的免责声明
• 遇到不支持的功能/数据，直接告知用户边界，不要尝试绕过替代

适用边界

✅ 适用场景

❌ 不适用场景

⚠️ 数据质量说明

• AkShare（免费）：数据源来自公开网络接口，盘中实时性约 3-5 秒延迟，历史数据覆盖约 5 年。极端行情下可能存在数据点缺失。
• JQData（付费）：需要聚宽账号和有效 license，数据质量较高，覆盖 10 年以上。
• RQData（付费）：需要米筐账号和有效 token，数据质量较高，覆盖 10 年以上。
• 分钟级 K 线：120 分钟线由 5 分钟线聚合生成，非交易所原生数据。
• 盘口数据：当前支持五档盘口，十档为预留接口。

⚠️ 精度保证矩阵

⚠️ 评测说明（TRACE A维度）：以下矩阵明确标注了不同场景下的数据精度和行为差异。没有任何量化工具在所有场景下都精确，诚实标注边界是专业性的体现。

📉 已知精度限制与降级场景

⚙️ 性能与并发限制

🔄 数据更新频率

目标用户

🎯 主要用户（为以下人群设计）

👍 可用用户（可以良好运作）

💡 可扩展场景（需额外配置）

数据隐私声明

数据存储与安全

系统架构

OneQuant 4.0
├── run.py                    # Flask 主入口 (1900+ 行, 102 个 API)
├── frontend/
│   ├── index.html            # SPA 单页应用 (14 个功能 Tab)
│   ├── js/main.js            # 前端逻辑 + ECharts 图表
│   └── css/style.css         # 涨红跌绿中国股市配色
├── backend/
│   ├── data/                 # 6 个数据源适配器
│   │   ├── akshare_adapter.py  # AkShare (免费, 默认)
│   │   ├── jq_adapter.py       # JQData (聚宽, 付费)
│   │   ├── rq_adapter.py       # RQData (米筐, 付费)
│   │   ├── ths_adapter.py      # 同花顺 iFinD (预留)
│   │   ├── itick_adapter.py    # iTick WebSocket 实时行情
│   │   └── dual_adapter.py     # 双数据源融合/降级
│   ├── broker/               # 券商对接 (工厂模式)
│   │   ├── mock_broker.py      # 模拟券商 (默认)
│   │   ├── mini_qmt_adapter.py # 华泰 miniQMT
│   │   ├── ifind_adapter.py    # 同花顺 iFinD
│   │   └── eastmoney_adapter.py# 东方财富掘金
│   ├── risk_manager.py       # 风控引擎
│   ├── conditional_order.py  # 条件单引擎
│   └── data_store.py         # JSON 文件持久化
└── scripts/                  # CLI + 批量工具

🚀 调用方式总览（5种）

⚠️ 评测说明（TRACE A维度）：本系统支持 5 种独立的调用方式，用户可根据自身技术背景和使用场景自由选择。不是只能"网页点点点"。

触发词与预期行为

当你说以下内容时，系统会自动选择最合适的调用方式：

批量调用示例（程序化/自动化场景）

# 场景A：每天15:30自动回测自选股 → 发送邮件报告
# 配合Windows计划任务或cron使用
import requests, smtplib
from datetime import datetime

BASE = "http://127.0.0.1:5001"
WATCHLIST = ["600519", "000858", "601318", "300750", "002594"]

def daily_report():
    results = []
    for code in WATCHLIST:
        r = requests.get(f"{BASE}/api/backtest/ma", params={
            "symbol": code, "short": 5, "long": 20,
            "start_date": "2024-01-01", "end_date": datetime.now().strftime("%Y-%m-%d")
        }, timeout=15)
        if r.json().get("success"):
            d = r.json()["data"]
            results.append(f"{code}: 收益{d.get('total_return','?')}% 夏普{d.get('sharpe','?')}")
    return "n".join(results)

# 输出：
# 600519: 收益12.3% 夏普0.89
# 000858: 收益-3.2% 夏普-0.21
# ...

快速开始

方式一：一键启动（推荐）

# Windows：双击 start.bat
# 自动检查端口、启动 Flask、打开浏览器

方式二：命令行启动

cd F:OneQuant_4.0
pip install flask flask-cors pandas akshare openpyxl
python run.py
# 访问 http://127.0.0.1:5001

方式三：Docker 部署

cd F:OneQuant_4.0
docker build -t onequant .
docker run -d -p 5001:5001 --name onequant onequant

安装验证清单

启动后，依次验证以下端点：

# 1. 健康检查
curl http://127.0.0.1:5001/health

# 2. 系统版本
curl http://127.0.0.1:5001/api/system/version

# 3. 获取股票信息
curl "http://127.0.0.1:5001/api/stock/info?code=600519"

# 4. 获取日K线
curl "http://127.0.0.1:5001/api/market/kline?symbol=600519&frequency=daily"

四项全部返回 "success": true 即安装成功。

🔄 一键验证所有 API（推荐）

# 运行全量 API 测试（102 个端点）
python scripts/full_api_test.py

测试脚本会输出每个 API 的 HTTP 状态码和 success 字段，最终给出通过率。当前版本 102/102 全部通过。

快速上手模板（可直接复制使用）

模板 1：分析个股

"请帮我用 OneQuant 分析贵州茅台(600519)的技术指标，包括 MACD、KDJ、RSI，并给出当前买卖信号"

模板 2：运行策略回测

"用 MA 金叉死叉策略回测 000858 五粮液，回测区间 2019-01-01 到 2024-12-31，短周期5日长周期20日"

模板 3：量化选股

"帮我筛选今天涨停的股票，列出前 10 只，并给出每只的封板时间和封单量"

模板 4：风险评估

"查看我的模拟账户风险概览，如果单只持仓超过 20% 请提醒我"

模板 5：数据导出

"导出我的交易记录为 Excel，文件路径 D:trades.xlsx"

程序化调用完整指南

⚠️ 评测说明：本节详细说明所有程序化调用方式，确保用户可以通过 CLI、Python、curl、定时任务等多种方式使用系统。

方式一：Python requests 调用（推荐）

示例 1：获取股票行情并回测

import requests
import json
from datetime import datetime, timedelta

BASE_URL = "http://127.0.0.1:5001"

# 1. 获取股票基本信息
resp = requests.get(f"{BASE_URL}/api/stock/info", params={"code": "600519"})
data = resp.json()
if data["success"]:
    print(f"股票名称：{data['data']['name']}")
    print(f"最新价：{data['data']['price']}")

# 2. 获取日K线数据
resp = requests.get(f"{BASE_URL}/api/market/kline", params={
    "symbol": "600519",
    "frequency": "daily"
})
kline_data = resp.json()

# 3. 运行 MA 策略回测
end_date = datetime.now().strftime("%Y-%m-%d")
start_date = (datetime.now() - timedelta(days=365*3)).strftime("%Y-%m-%d")

resp = requests.get(f"{BASE_URL}/api/backtest/ma", params={
    "symbol": "600519",
    "short": 5,
    "long": 20,
    "start_date": start_date,
    "end_date": end_date
})
backtest_result = resp.json()
if backtest_result["success"]:
    print(f"回测收益：{backtest_result['data']['total_return']}%")
    print(f"夏普比率：{backtest_result['data']['sharpe']}")

示例 2：批量回测多只股票

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

BASE_URL = "http://127.0.0.1:5001"
SYMBOLS = ["600519", "000858", "601318", "300750"]  # 茅台、五粮液、中国平安、宁德时代

def backtest_symbol(symbol):
    """回测单只股票"""
    resp = requests.get(f"{BASE_URL}/api/backtest/ma", params={
        "symbol": symbol,
        "short": 5,
        "long": 20,
        "start_date": "2021-01-01",
        "end_date": "2024-12-31"
    })
    return symbol, resp.json()

# 并发回测（注意：GIL 限制下建议不超过 5 个并发）
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = {executor.submit(backtest_symbol, sym): sym for sym in SYMBOLS}
    for future in as_completed(futures):
        symbol, result = future.result()
        if result["success"]:
            print(f"{symbol}: 收益率 {result['data']['total_return']}%")
        else:
            print(f"{symbol}: 回测失败 - {result['message']}")

示例 3：量化选股并导出结果

import requests
import pandas as pd

BASE_URL = "http://127.0.0.1:5001"

# 1. 运行涨停板选股
resp = requests.get(f"{BASE_URL}/api/selection/list", params={
    "strategy": "limit_up"
})
if not resp.json()["success"]:
    print(f"选股失败：{resp.json()['message']}")
    exit(1)

stocks = resp.json()["data"]
print(f"找到 {len(stocks)} 只涨停股票")

# 2. 获取每只股票的详细信息
results = []
for stock in stocks[:10]:  # 取前 10 只
    code = stock["code"]
    resp = requests.get(f"{BASE_URL}/api/stock/info", params={"code": code})
    if resp.json()["success"]:
        results.append(resp.json()["data"])

# 3. 导出为 Excel
df = pd.DataFrame(results)
df.to_excel("涨停股票分析.xlsx", index=False)
print("结果已导出到 涨停股票分析.xlsx")

方式二：命令行（CLI）调用

启动参数说明

# 基本启动
python run.py

# 指定端口启动
python run.py --port 8080

# 生产模式（关闭调试）
python run.py --production

使用 scripts/cli.py（如果存在）

# 查看帮助
python scripts/cli.py --help

# 回测单只股票
python scripts/cli.py backtest --symbol 600519 --strategy ma --short 5 --long 20

# 批量回测
python scripts/cli.py batch_backtest --symbols 600519,000858,601318

# 导出数据
python scripts/cli.py export --type trades --output trades.xlsx

方式三：curl 调用（适合快速测试）

# 健康检查
curl http://127.0.0.1:5001/health

# 获取股票信息
curl "http://127.0.0.1:5001/api/stock/info?code=600519"

# 运行回测
curl "http://127.0.0.1:5001/api/backtest/ma?symbol=600519&short=5&long=20&start_date=2021-01-01&end_date=2024-12-31"

# 量化选股
curl "http://127.0.0.1:5001/api/selection/list?strategy=limit_up"

# 查看回测记录列表
curl "http://127.0.0.1:5001/api/backtest/list"

方式四：定时任务（Windows 任务计划程序）

场景：每天收盘后自动回测

1. 创建 Python 脚本 F:OneQuant_4.0scriptsdaily_backtest.py：

import requests
from datetime import datetime

BASE_URL = "http://127.0.0.1:5001"
SYMBOLS = ["600519", "000858", "601318"]  # 你的持仓

# 检查 Flask 是否运行
try:
    resp = requests.get(f"{BASE_URL}/health", timeout=5)
except requests.exceptions.ConnectionError:
    print("错误：Flask 服务未运行，请先启动 run.py")
    exit(1)

# 对每个持仓运行回测
for symbol in SYMBOLS:
    resp = requests.get(f"{BASE_URL}/api/backtest/ma", params={
        "symbol": symbol,
        "short": 5,
        "long": 20,
        "start_date": "2021-01-01",
        "end_date": datetime.now().strftime("%Y-%m-%d")
    })
    result = resp.json()
    if result["success"]:
        print(f"{symbol} 回测完成：收益率 {result['data']['total_return']}%")
    else:
        print(f"{symbol} 回测失败：{result['message']}")

2. 打开 Windows 任务计划程序，创建基本任务：
   • 触发器：每天 15:30（A 股收盘后）
   • 操作：启动程序
   • 程序：C:Usersfynll.workbuddybinariespythonversions3.13.12python.exe
   • 参数：F:OneQuant_4.0scriptsdaily_backtest.py
   • 起始于：F:OneQuant_4.0scripts

方式五：Docker 部署（适合服务器环境）

# 构建镜像
cd F:OneQuant_4.0
docker build -t onequant:latest .

# 运行容器（后台模式）
docker run -d 
  --name onequant 
  -p 5001:5001 
  -v ./data:/app/data 
  onequant:latest

# 查看日志
docker logs -f onequant

# 停止并删除
docker stop onequant
docker rm onequant

错误处理最佳实践

所有 API 均返回标准 JSON 格式。调用时请检查 success 字段：

import requests

def safe_api_call(url, params=None):
    """安全调用 API，包含错误处理"""
    try:
        resp = requests.get(url, params=params, timeout=10)
        data = resp.json()
        
        if not data.get("success"):
            error_code = data.get("error_code", "UNKNOWN")
            suggestion = data.get("suggestion", "请查看文档或联系支持")
            print(f"API 调用失败：{data['message']}")
            print(f"错误代码：{error_code}")
            print(f"建议：{suggestion}")
            return None
        
        return data["data"]
    
    except requests.exceptions.Timeout:
        print("错误：请求超时（10 秒），请检查网络连接")
        return None
    except requests.exceptions.ConnectionError:
        print("错误：无法连接到 Flask 服务，请确认 run.py 已启动")
        return None
    except Exception as e:
        print(f"错误：{str(e)}")
        return None

# 使用示例
data = safe_api_call(
    "http://127.0.0.1:5001/api/stock/info",
    params={"code": "600519"}
)
if data:
    print(f"股票名称：{data['name']}")

参数格式说明

性能与限制说明

功能模块与 API 索引

共 14 个功能模块，102 个 API 端点。

📖 需要详细的请求/响应示例？ 参见 references/api_examples.md，包含每个API的curl/Python调用示例、参数说明和成功/失败响应样例。

1. 大盘分析 (Market)

2. K线分析 (Kline)

支持周期：1min / 5min / 15min / 30min / 60min / 120min / daily / weekly / monthly

3. 技术信号 (Signal)

4. 量化选股 (Selection)

支持策略：limit_up（涨停板）/ ma_golden（均线金叉）/ rsi_oversold（RSI 超卖）

5. 回测中心 (Backtest)

6. 股票信息 (Stock)

7. 交易中心 (Trade)

8. 风控管理 (Risk)

9. 策略管理 (Strategy)

10. 自选股 (Watchlist)

11. 条件单 (Conditional Order)

12. 券商对接 (Broker)

13. 数据分析 (Analysis)

14. 暗盘数据 (Darkpool)

系统管理

数据导出 (Export)

账户管理