猎户座量化交易部署全流程指南
引言：小王的量化部署之路
小王是一名程序员，平时对股票投资很感兴趣，但工作繁忙，很难有时间盯盘。听朋友说猎户座的AI量化交易系统能自动完成选股、买卖决策，而且胜率稳定在60%-70%，他决定试一试。
然而，当他真正开始动手时，才发现"量化交易"并没有想象中那么简单——需要申请权限、架设服务器、安装券商软件、部署策略代码、配置网络白名单……每一步都涉及不同的系统和操作。小王花了两周时间摸索，踩了不少坑，才终于把整个流程打通。
为了让后来的量化爱好者不再走弯路，小王把自己的部署经验整理成这份文档。只要你按照以下步骤操作，即使是零基础的小白，也能在一天内完成全流程部署，让AI替你自动交易。
本指南以国信证券iQuant量化终端为例，如果你使用的是其他券商，流程大同小异，可以参考本文档的思路。
一、部署全景图
在开始之前，先了解整个量化交易系统的架构，这对理解每一步的作用非常重要：
┌─────────────────────────────────────────────────────────────────┐
│ 猎户座AI决策系统 │
│ (deephunt.com.cn) │
│ ┌─────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ 交易技能Skills│ │ 账号配置 │ │ 安全中心(IP白名单) │ │
│ └─────────────┘ └──────────────┘ └──────────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ API接口 (getbuystock / getsellstock 等) │ │
│ └──────────────────────────┬───────────────────────────────┘ │
└──────────────────────────────┼──────────────────────────────────┘
│ HTTPS POST请求
▼
┌─────────────────────────────────────────────────────────────────┐
│ Windows云端服务器 (阿里云ECS) │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ 券商量化交易软件 (iQuant) │ │
│ │ ┌────────────────────────────────────────────────────┐ │ │
│ │ │ 猎户座量化策略代码 (Python) │ │ │
│ │ │ 1. 定时POST调用猎户座API获取买卖信号 │ │ │
│ │ │ 2. 解析API返回的JSON股票列表 │ │ │
│ │ │ 3. 通过iQuant接口下单/撤单 │ │ │
│ │ │ 4. 成交后POST回传交割指令到猎户座 │ │ │
│ │ └────────────────────────────────────────────────────┘ │ │
│ └──────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ 券商交易系统 (实盘账户) │ │
│ └──────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘核心逻辑：猎户座负责"告诉你要买什么/卖什么"，券商量化软件负责"执行买卖操作"。两者通过API接口以POST方式通信，运行在云端服务器上实现无人值守。
部署流程总览（共4大步骤）：
| 步骤 | 内容 | 预计耗时 | 关键产出 |
|---|---|---|---|
| 第一步 | 申请开通量化交易权限 | 1-3个工作日 | 猎户座量化权限 + 券商量化权限 |
| 第二步 | 开通云端服务器 | 30分钟 | 一台阿里云Windows服务器 |
| 第三步 | 部署券商量化交易软件 | 1-2小时 | iQuant安装并登录实盘账号 |
| 第四步 | 部署策略 + 配置连通 | 1-2小时 | 策略代码运行 + API连通 + 自动运行 |
二、应用场景
场景一：上班族的无人值守量化交易
小张是一名996程序员，白天根本没时间看盘。他按照本指南部署好量化交易后，每天早上9:25系统自动启动，AI根据他设定的交易技能Skills筛选买卖信号，自动下单。下午15:00收盘后自动关闭。他只需要偶尔打开猎户座App查看历史收益即可，真正实现了"躺着赚钱"。
场景二：专栏作家的量化教程演示
李老师是金融量化专栏作家，他需要向读者展示一套完整的量化交易部署方案。他按照本指南在阿里云服务器上完整走了一遍流程，截图记录每一步，形成了一篇万字教程，帮助数百名读者成功部署了量化交易系统。
场景三：多策略并行运行
王先生有3套不同的交易策略，他在同一台服务器上部署了3个iQuant实例，分别对应不同的猎户座交易技能Skills，在账号配置中为每个市场设置独立参数，实现了多策略并行运行，分散风险的同时也提高了资金利用率。
三、第一步：申请开通量化交易权限
在开始技术部署之前，必须先完成权限申请。这一步需要同时在猎户座平台和券商两端操作。
3.1 在猎户座申请开通量化交易权限
操作路径：登录猎户座平台 → 量化交易 → 申请开通
操作步骤：
- 打开浏览器，访问猎户座平台
http://www.deephunt.com.cn - 使用你的账号登录（如未注册请先注册）
- 在左侧导航栏找到"量化交易"，点击进入
- 点击"申请开通"页面
- 阅读并同意《量化交易服务协议》
- 提交申请，等待系统审核
注意事项：
- 猎户座的量化交易权限通常需要付费开通，请确认你的账号已购买相应的服务套餐
- 审核通过后，你可以在"量化交易"页面看到完整的量化功能菜单，详情可参考量化交易操作指南
- 如果看不到"量化交易"菜单，说明你的账号尚未开通量化权限，请联系客服
3.2 联系券商申请开通量化交易权限
以国信证券为例，开通量化实盘交易权限的流程如下。以下为完整的申请报备流程示范：
参考文档：点击下载 4-国信证券量化实盘权限申请报备实例.docx（在新窗口打开后可下载）
📄 国信证券量化实盘权限申请报备实例（完整内容）
================================================================
国信证券量化实盘权限申请报备实例
================================================================
一、申请前准备
---------------
1. 确认已在国信证券开立普通证券账户
- 账户类型：沪深A股账户
- 账户状态：正常，无冻结、无异常限制
2. 确认账户满足开通条件
- 开户时间：满3个月以上
- 风险测评：已完成且为积极型或进取型
- 交易经验：有一定的股票交易经验
3. 准备材料清单
- 身份证正反面照片（清晰）
- 本人手持身份证照片
- 已签署的《量化交易风险揭示书》
- 已填写的《量化交易权限申请表》
二、联系客户经理
---------------
方式一：直接联系开户时的客户经理
- 通过微信/电话联系
- 说明："我需要开通量化交易（iQuant）实盘权限"
方式二：拨打国信证券客服电话
- 客服电话：95536
- 转人工服务
- 说明："我需要开通量化交易权限，请帮我转接相关部门"
方式三：通过国信证券APP
- 打开国信证券APP
- 进入"我的" → "业务办理"
- 搜索"量化交易"或"iQuant"
- 在线提交申请
三、填写《量化交易权限申请表》
-----------------------------
以下为申请表需要填写的关键信息：
┌─────────────────────────────────────────────────────────┐
│ 国信证券量化交易权限申请表 │
├─────────────────────────────────────────────────────────┤
│ │
│ 申请人姓名：___________ 身份证号：________________ │
│ │
│ 资金账号：___________ 联系电话：________________ │
│ │
│ 申请类型： │
│ □ 量化交易（iQuant）实盘权限 │
│ □ 量化交易（iQuant）模拟权限 │
│ │
│ 策略类型： │
│ □ 股票多头策略 │
│ □ 股票多空策略 │
│ □ 其他：________________ │
│ │
│ 策略描述（简要说明）： │
│ ________________________________________________ │
│ ________________________________________________ │
│ ________________________________________________ │
│ │
│ 预计日均交易频率： │
│ □ 低频（日均10笔以内） │
│ □ 中频（日均10-50笔） │
│ □ 高频（日均50笔以上） │
│ │
│ 预计日均交易金额：___________ 万元 │
│ │
│ 申请人签名：___________ 日期：___________ │
│ │
└─────────────────────────────────────────────────────────┘
四、签署《量化交易风险揭示书》
-----------------------------
风险揭示书核心内容（需仔细阅读并签字确认）：
1. 量化交易存在策略失效风险，历史回测收益不代表未来表现
2. 程序化交易可能出现系统故障、网络延迟等技术风险
3. 市场极端行情下，量化策略可能产生超出预期的亏损
4. 投资者需自行承担量化交易的全部风险和损失
5. 券商不对量化策略的收益做任何承诺或保证
6. 投资者应确保策略代码的合法合规性，不得用于市场操纵
五、提交材料清单
---------------
请将以下材料通过邮件或微信发送给客户经理：
□ 身份证正反面照片（JPG格式，清晰可见）
□ 本人手持身份证照片（JPG格式）
□ 已签署的《量化交易风险揭示书》（PDF格式）
□ 已填写的《量化交易权限申请表》（PDF格式）
□ 策略简要说明文档（Word格式，可选）
六、审核流程与时间节点
---------------------
步骤1：提交材料 → 客户经理初审（1个工作日内）
步骤2：初审通过 → 风控部门复审（1-2个工作日）
步骤3：复审通过 → 系统开通权限（当天生效）
步骤4：权限开通 → 客户经理通知并发送iQuant下载链接
总计：约1-3个工作日
七、审核通过后的操作
-------------------
1. 收到审核通过的短信/电话通知
2. 客户经理提供iQuant下载链接和安装包
3. 获取iQuant登录账号和初始密码
4. 首次登录后请立即修改密码
5. 登录iQuant验证实盘交易权限是否正常
八、常见驳回原因及应对
---------------------
驳回原因1：风险测评等级不够
应对：重新进行风险测评，选择积极型或进取型
驳回原因2：策略描述过于简单
应对：补充策略逻辑说明，包括选股条件、风控措施等
驳回原因3：账户交易经验不足
应对：先进行一段时间的模拟交易，积累交易记录后再申请
驳回原因4：材料不完整
应对：按照清单逐项检查，确保所有材料齐全
================================================================常见问题：
- Q：没有客户经理怎么办？ A：拨打国信证券官方客服电话95536，转人工服务，说明需要开通量化交易权限，客服会帮你转接相关部门。
- Q：开户不满6个月可以申请吗？ A：不同券商要求不同，国信证券一般要求开户满3个月以上，具体以券商最新政策为准。
- Q：模拟交易需要申请吗？ A：不需要。iQuant自带模拟交易功能，可以直接使用。猎户座的模拟交易只需要在账号配置中选择"模拟交易"模式即可，无需连接券商。
四、第二步：开通云端服务器
量化交易需要7×24小时运行，用个人电脑显然不现实——关机、断网、休眠都会导致交易中断。因此，你需要一台云端服务器。
4.1 选择服务器配置
推荐配置：
| 配置项 | 最低要求 | 推荐配置 |
|---|---|---|
| 操作系统 | Windows Server 2016/2019/2022 | Windows Server 2022 中文版 |
| CPU | 2核 | 4核 |
| 内存 | 4GB | 8GB |
| 硬盘 | 40GB SSD | 80GB SSD |
| 带宽 | 1Mbps | 3-5Mbps |
为什么选择Windows服务器？
- 国信证券iQuant量化终端目前仅支持Windows系统
- 券商提供的量化交易软件大多基于Windows开发
推荐云服务商：
- 阿里云ECS（推荐，本指南以阿里云为例）
- 腾讯云轻量应用服务器（性价比高）
- 华为云弹性云服务器（有专门的金融云服务）
价格参考：阿里云ECS 4核8G配置，约150-200元/月，足以满足量化交易需求。
4.2 购买并初始化阿里云服务器
第一步：注册阿里云账号
- 访问阿里云官网
https://www.aliyun.com - 点击右上角"免费注册"
- 按提示完成实名认证（需要身份证信息，审核约1-2小时）
- 建议先领取新用户优惠券，可享受首年折扣
第二步：购买ECS服务器
- 登录阿里云控制台，进入"弹性计算" → "云服务器ECS"
- 点击"创建实例"
- 按以下配置选择：
| 配置项 | 选择 |
|---|---|
| 付费方式 | 包年包月（长期使用更优惠）或按量付费（测试用） |
| 地域 | 选择离你最近的区域（如华东1-杭州、华东2-上海、华南1-深圳） |
| 实例规格 | 4核8GB（ecs.c6.xlarge 或 ecs.g6.xlarge） |
| 镜像 | Windows Server 2022 中文版 |
| 系统盘 | SSD云盘 80GB |
| 带宽 | 按固定带宽 3-5Mbps |
| 安全组 | 默认安全组（开放3389端口用于远程桌面） |
- 设置登录密码（请务必牢记，建议包含大小写字母、数字和特殊字符）
- 确认订单，完成支付
第三步：远程登录服务器
- 在阿里云控制台找到你购买的ECS实例
- 点击实例右侧的"远程连接"
- 选择"Workbench远程连接"（推荐）或下载RDP文件
- 输入服务器公网IP、用户名（Administrator）和密码
- 成功登录后，你会看到Windows Server桌面
提示：Mac用户可以使用 Microsoft Remote Desktop 客户端（在App Store免费下载），连接时输入服务器IP、用户名和密码即可。
第四步：基础配置
登录服务器后，进行以下基础配置：
关闭IE增强安全配置（否则无法正常浏览网页）：
- 打开"服务器管理器"
- 点击"本地服务器"
- 找到"IE增强的安全配置"，点击"启用"
- 将"管理员"和"用户"都设置为"关闭"
安装Google Chrome浏览器（方便后续操作）：
- 打开IE浏览器，访问
https://www.google.cn/chrome/ - 下载并安装Chrome
- 打开IE浏览器，访问
设置服务器永不休眠：
- 右键点击桌面 → 个性化 → 锁屏界面 → 屏幕保护程序设置
- 屏幕保护程序选择"无"
- 点击"更改电源设置" → 选择"高性能"
- 将"关闭显示器"和"使计算机进入睡眠状态"都设置为"从不"
配置自动登录（防止重启后需要手动输入密码）：
- 按
Win + R，输入netplwiz，回车 - 取消勾选"要使用本计算机，用户必须输入用户名和密码"
- 点击"应用"，输入你的管理员密码确认
- 这样服务器重启后会自动登录到桌面
- 按
配置阿里云安全组（确保远程桌面可访问）：
- 在阿里云控制台 → 安全组 → 配置规则
- 确认入方向已开放3389端口（Windows远程桌面）
- 建议将3389端口的授权对象限制为你的本地IP，提高安全性
五、第三步：安装券商量化交易软件
5.1 下载并安装iQuant量化终端
参考文档：点击下载 1-国信证券iQuant量化终端部署核心流程.docx（在新窗口打开后可下载）
📄 国信证券iQuant量化终端部署核心流程（完整内容）
================================================================
国信证券iQuant量化终端部署核心流程
================================================================
一、iQuant终端简介
------------------
iQuant是国信证券推出的量化交易终端，支持：
- Python策略编写与回测
- 模拟交易与实盘交易
- 多账户管理
- 定时任务与自动运行
- 交易日志与绩效分析
二、获取iQuant安装包
--------------------
方式一：客户经理提供
- 联系国信证券客户经理
- 获取iQuant最新版下载链接
- 下载安装包（通常为 .exe 格式，约200-500MB）
方式二：国信证券官网下载
- 访问 https://www.guosen.com.cn
- 进入"量化交易"或"iQuant"专区
- 下载最新版本
三、安装iQuant终端
------------------
步骤1：双击安装包开始安装
- 如果在服务器上，先将安装包上传到服务器
- 可以直接在服务器上下载Chrome，然后下载安装包
步骤2：安装向导设置
- 安装路径：建议使用默认路径 C:\iQuant\
- 勾选"创建桌面快捷方式"
- 其他选项保持默认
步骤3：完成安装
- 安装过程约3-5分钟
- 安装完成后桌面出现iQuant图标
四、首次登录iQuant
------------------
步骤1：启动iQuant
- 双击桌面iQuant图标
- 等待软件初始化（首次可能需要1-2分钟）
步骤2：输入登录信息
账号：________________
密码：________________
验证码：____________ 点击获取
交易模式：○ 实盘 ○ 模拟
登 录
- 账号：由客户经理提供的iQuant账号
- 密码：初始密码（首次登录后建议修改）
- 交易模式：先选择"模拟"测试，确认无误后再选"实盘"
步骤3：设置交易密码
- 首次登录可能要求设置交易密码
- 交易密码用于下单时的二次验证
- 建议与登录密码不同
步骤4：验证登录成功
- 登录后应能看到主界面
- 左侧菜单栏包含：策略管理、账户信息、交易日志等
五、iQuant主界面介绍
--------------------
主界面布局：
iQuant 量化交易终端 ─ □ ×
─────────────────────────────────
┌──────────┬──────────────────────────────────────────────┐
│ 菜单栏 │ 主工作区 │
│ ● 策略管理│ │
│ ● 账户信息│ │
│ ● 持仓查询│ │
│ ● 交易日志│ │
│ ● 参数设置│ │
│ ● 系统设置│ │
└──────────┴──────────────────────────────────────────────┘
状态栏：已连接 | 账户：xxxxxx | 模式：实盘 | 时间：09:30
核心功能模块说明：
1. 策略管理
- 新建策略：创建Python策略代码
- 导入策略：从文件导入已有策略
- 策略回测：对策略进行历史数据回测
- 运行策略：启动策略实时运行
2. 账户信息
- 总资产：账户总市值
- 可用资金：可用于买入的资金
- 持仓市值：当前持仓股票市值
- 当日盈亏：当天盈亏金额
3. 持仓查询
- 股票代码、名称
- 持仓数量、成本价
- 当前价格、盈亏比例
4. 交易日志
- 下单记录：买入/卖出订单
- 成交记录：已成交的订单
- 撤单记录：已撤销的订单
- 策略日志：策略运行日志
六、iQuant系统设置
------------------
建议完成以下系统设置：
1. 交易设置
- 默认交易数量：根据资金量设置
- 默认交易价格：选择"限价"或"市价"
- 下单确认：建议开启"下单前确认"
2. 通知设置
- 成交通知：开启
- 策略异常通知：开启
- 通知方式：弹窗 + 声音
3. 日志设置
- 日志保存路径：使用默认
- 日志保留天数：建议30天
- 自动清理：开启
================================================================第一步：获取安装包
- 国信证券客户经理会提供iQuant的下载链接
- 或者从国信证券官网的量化交易专区下载
- 下载文件通常是一个
.exe安装包
第二步：安装iQuant
- 在服务器上双击安装包，开始安装
- 按照安装向导提示操作，建议使用默认安装路径（如
C:\iQuant\） - 安装完成后，桌面会出现iQuant的快捷方式图标
第三步：登录iQuant
- 双击桌面上的iQuant图标，启动软件
- 输入你的iQuant账号和密码（由券商客户经理提供）
- 选择"实盘交易"模式
- 点击"登录"
注意：首次登录可能需要设置交易密码，请根据提示操作。交易密码是独立于登录密码的，用于下单时的二次验证。
第四步：验证登录成功
登录成功后，你应该能看到：
- 账户信息页面，显示你的可用资金、持仓等信息
- 策略管理页面，用于后续部署策略代码
- 交易日志页面，用于查看交易记录
5.2 熟悉iQuant基本操作
在部署策略之前，建议先熟悉iQuant的几个核心功能：
| 功能模块 | 用途 | 说明 |
|---|---|---|
| 策略管理 | 编写/导入/运行策略 | 策略代码在这里编写和运行 |
| 账户信息 | 查看资金和持仓 | 实时显示账户状态 |
| 交易日志 | 查看交易记录 | 每一笔下单和成交都会记录 |
| 参数设置 | 配置策略参数 | 设置策略的运行参数 |
六、第四步：部署策略 + 配置连通
这是最关键的一步，需要将猎户座的量化策略代码部署到iQuant中，并配置好API连通性。
6.1 获取猎户座API账号信息
在部署策略代码之前，需要先获取你的API认证信息。详细配置说明可参考账号配置和安全中心文档。
第一步：获取AppID
- 登录猎户座平台
- 进入"量化交易" → "账号配置"
- 在"全局控制区"找到你的 AppID（系统自动生成，格式如
hunt6da290c4c4b93847） - 复制保存这个AppID
第二步：获取AppSecret
- 进入"量化交易" → "安全中心"
- 如果尚未创建AppSecret，点击"创建秘钥"按钮
- 点击"拷贝"按钮，将AppSecret复制到安全的地方
- 重要：AppSecret只显示一次，请务必妥善保存！如果丢失，需要重新生成
6.2 配置IP白名单
为了让iQuant所在的服务器能够访问猎户座API，需要将服务器的公网IP加入白名单。详细配置可参考安全中心文档。
第一步：获取服务器公网IP
在服务器上打开Chrome浏览器，访问 https://whatismyipaddress.com/，页面会显示你的公网IP地址（格式如 123.456.789.123）。
也可以在服务器上打开命令提示符（CMD），输入以下命令：
curl ifconfig.me第二步：在猎户座配置IP白名单
- 登录猎户座平台
- 进入"量化交易" → "安全中心"
- 点击"修改"按钮
- 在"IP白名单"文本框中，输入你的服务器公网IP（每行一个IP）
- 点击"保存"
- 系统提示保存成功后，配置立即生效
注意：如果你的阿里云服务器IP可能会变化（如使用弹性公网IP），建议使用固定公网IP，或者在安全组中绑定弹性IP。
6.3 部署策略代码到iQuant
参考文档：点击下载 2-国信证券iQuant量化代码V8.0-支持卖单成功后继续提取新的买单持续高效交易-标准代码.docx（在新窗口打开后可下载）
📄 iQuant量化交易标准代码 V8.0（完整内容）
python
#coding:gbk
"""
================================================================
猎户座量化交易策略 - iQuant版本 V8.0
功能：支持卖单成功后继续提取新的买单，持续高效交易
适配：国信证券iQuant量化终端
作者：猎户座量化交易团队
================================================================
策略核心逻辑：
1. 定时POST调用猎户座API获取买卖信号
2. 解析API返回的JSON股票列表
3. 根据信号执行买入/卖出操作
4. 成交后POST回传交割指令到猎户座
5. 卖单成功后立即提取新的买单信号，持续高效交易
6. 自动止损止盈管理
"""
import json
import time
from datetime import datetime
import urllib.request
import urllib.error
import urllib.parse
import threading
# ==================== 配置区域 ====================
ORION_CONFIG = {
"appid": "你的AppID", # 替换为猎户座"账号配置"中的AppID
"appsecret": "你的AppSecret", # 替换为猎户座"安全中心"中的AppSecret
"cid": 1, # 1=A股, 2=港股, 3=美股
"result_column_ids": "4,5",
"api_get_buy": "http://www.deephunt.com.cn/api/getbuystock",
"api_get_sell": "http://www.deephunt.com.cn/api/getsellstock",
"api_exec_buy": "http://www.deephunt.com.cn/api/executeuserbuytrade",
"api_exec_sell": "http://www.deephunt.com.cn/api/executeuserselltrade",
}
TRADE_ACCOUNT = "你的账号" # 国信量化实盘账号
ORDER_CONFIG = {
"op_type_buy": 23,
"op_type_sell": 24,
"order_type": 1101,
"pr_type_limit": 11,
"quick_trade": 2,
"strategy_name": "猎户座量化策略",
}
# ==================== 循环交易配置 ====================
# 卖出成功后，等待多少秒后重新获取买单
REFRESH_BUY_INTERVAL = 10 # 10秒
# =================================================
MAX_RETRY_COUNT = 10
RETRY_INTERVAL = 60
TEST_MODE = False
# ==================== 时间窗口配置 ====================
# 早盘时间窗口（9:30-9:40）- 执行买入+卖出
MORNING_HOUR = 9
MORNING_MIN_START = 30
MORNING_MIN_END = 40
# 收盘前时间窗口（14:55-14:59）- 只执行卖出
CLOSE_HOUR = 14
CLOSE_MIN_START = 55
CLOSE_MIN_END = 59
# 交易时段（9:30-15:00）
TRADE_START_HOUR = 9
TRADE_START_MIN = 30
TRADE_END_HOUR = 15
TRADE_END_MIN = 0
# =================================================
MARKET_MAPPING = {
"sz": ".SZ",
"sh": ".SH",
"bj": ".BJ",
"hk": ".HK",
"us": ".US",
}
class StrategyState:
def __init__(self):
self.last_morning_date = ""
self.last_close_date = ""
self.pending_orders = {} # key: instruction_id, value: 订单信息
self.pending_deals = {} # 待处理的成交信息
self.executed_instructions = set()
self.is_executing = False
self.last_check_time = 0
# ========== 新增：循环交易状态 ==========
self.last_buy_fetch_time = "" # 最后一次获取买单的时间
self.sell_triggered_buy_fetch = False # 是否因卖出触发了买单获取
self.pending_buy_fetch = False # 是否有待执行的买单获取任务
state = StrategyState()
# =================================================
def format_price(price, market="A"):
"""格式化价格"""
if price is None:
return 0.0
if market == "HK":
return round(price, 3)
elif market == "US":
return round(price, 2)
else:
return round(price, 2)
def convert_stock_code(orion_code):
"""转换股票代码格式"""
if not orion_code:
return None
orion_code = orion_code.strip().lower()
for prefix, suffix in MARKET_MAPPING.items():
if orion_code.startswith(prefix):
code_num = orion_code[len(prefix):]
return f"{code_num}{suffix}"
return orion_code
def get_market_from_code(orion_code):
"""根据股票代码前缀判断市场"""
if not orion_code:
return "A"
code_lower = orion_code.strip().lower()
if code_lower.startswith("hk"):
return "HK"
elif code_lower.startswith("us"):
return "US"
else:
return "A"
def is_trading_time():
"""判断当前是否在交易时间内"""
now = datetime.now()
current_hour = now.hour
current_minute = now.minute
current_weekday = now.weekday()
# 周末不交易
if current_weekday >= 5:
return False
# 检查是否在 9:30-15:00 之间
if current_hour < TRADE_START_HOUR or current_hour > TRADE_END_HOUR:
return False
if current_hour == TRADE_START_HOUR and current_minute < TRADE_START_MIN:
return False
if current_hour == TRADE_END_HOUR and current_minute >= TRADE_END_MIN:
return False
return True
def init(ContextInfo):
"""策略初始化"""
global state
ContextInfo.set_account(TRADE_ACCOUNT)
# 重置全局状态
state.last_morning_date = ""
state.last_close_date = ""
state.pending_orders = {}
state.pending_deals = {}
state.executed_instructions = set()
state.is_executing = False
state.last_check_time = 0
state.last_buy_fetch_time = ""
state.sell_triggered_buy_fetch = False
state.pending_buy_fetch = False
print("=" * 70)
print("猎户座-国信桥接策略已启动 (v8.0 - 循环交易模式)")
print(f"交易账号: {TRADE_ACCOUNT}")
print(f"AppID: {ORION_CONFIG['appid']}")
print(f"测试模式: {'开启' if TEST_MODE else '关闭'}")
print(f"启动时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("-" * 70)
print("【循环交易配置】")
print(f" 卖出成功后，{REFRESH_BUY_INTERVAL}秒后重新获取买单")
print("-" * 70)
print("【时间窗口配置】")
print(f" 早盘时间窗口 ({MORNING_HOUR:02d}:{MORNING_MIN_START:02d}-{MORNING_MIN_END:02d}): 买入 + 卖出")
print(f" 收盘前时间窗口 ({CLOSE_HOUR:02d}:{CLOSE_MIN_START:02d}-{CLOSE_MIN_END:02d}): 仅卖出")
print("-" * 70)
print("【交易时段】")
print(f" 9:30-15:00 (循环交易仅在此时间段内触发)")
print("=" * 70)
# 检查账号状态
try:
accounts = get_trade_detail_data(TRADE_ACCOUNT, 'stock', 'account')
if accounts:
for acc in accounts:
print(f"[账号状态] 可用资金: {getattr(acc, 'm_dAvailable', 'N/A')}")
else:
print("[警告] 无法获取账号信息，请确认账号已登录")
except Exception as e:
print(f"[账号检查] {str(e)}")
print("\n策略持续运行中，将在时间窗口内自动执行任务...")
print("=" * 70)
def _call_orion_api_post_form(url, params, max_retries=1):
"""调用猎户座API"""
for attempt in range(max_retries):
try:
request_data = {
"appid": ORION_CONFIG["appid"],
"appsecret": ORION_CONFIG["appsecret"],
}
if params:
request_data.update(params)
encoded_data = urllib.parse.urlencode(request_data).encode('utf-8')
print(f"[API请求] URL: {url}")
req = urllib.request.Request(
url,
data=encoded_data,
headers={
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': 'application/json',
},
method='POST'
)
with urllib.request.urlopen(req, timeout=15) as response:
response_data = response.read().decode('utf-8')
result = json.loads(response_data)
if result.get("code") == 1:
data = result.get("data")
if data is None:
return True, []
if isinstance(data, list):
return True, data
return True, []
else:
error_msg = result.get("message", "未知错误")
print(f"[API错误] {error_msg}")
return False, None
except Exception as e:
print(f"[API异常] {str(e)}")
if attempt == max_retries - 1:
return False, None
return False, None
def execute_buy_order(ContextInfo, instruction):
"""执行买入指令 - 使用remark传递指令ID"""
global state
orion_code = instruction.get("Code")
buy_shares = instruction.get("BuySum")
instruction_id = instruction.get("ID")
buy_price = instruction.get("BuyPrice")
if not orion_code:
print(f"[买入跳过] 指令{instruction_id}缺少股票代码")
return False
if not buy_shares or buy_shares <= 0:
print(f"[买入跳过] 指令{instruction_id}股数无效: {buy_shares}股")
return False
iquant_code = convert_stock_code(orion_code)
if not iquant_code:
print(f"[买入跳过] 指令{instruction_id}代码转换失败: {orion_code}")
return False
buy_shares = int(buy_shares / 100) * 100
if buy_shares <= 0:
print(f"[买入跳过] 指令{instruction_id}股数无效: {buy_shares}股")
return False
market = get_market_from_code(orion_code)
formatted_price = format_price(buy_price, market)
print(f"\n[买入执行] ID={instruction_id}")
print(f" 代码: {orion_code} -> {iquant_code}")
print(f" 价格: {formatted_price}, 股数: {buy_shares}")
try:
state.pending_orders[str(instruction_id)] = {
"instruction_id": instruction_id,
"type": "buy",
"stock_code": iquant_code,
"expected_shares": buy_shares,
"expected_price": formatted_price,
"order_time": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"status": "submitted",
"order_sys_id": None
}
remark = f"INSTR_{instruction_id}"
passorder(
23, 1101, TRADE_ACCOUNT, iquant_code, 11, formatted_price,
buy_shares, ORDER_CONFIG["strategy_name"], ORDER_CONFIG["quick_trade"],
remark, ContextInfo
)
print(f"[买入提交] 指令{instruction_id}已提交，remark={remark}")
return True
except Exception as e:
print(f"[买入异常] {str(e)}")
if str(instruction_id) in state.pending_orders:
del state.pending_orders[str(instruction_id)]
return False
def execute_sell_order(ContextInfo, instruction):
"""执行卖出指令 - 使用remark传递指令ID"""
global state
orion_code = instruction.get("Code")
sell_shares = instruction.get("BuySum")
instruction_id = instruction.get("ID")
sell_price = instruction.get("BuyPrice")
if not orion_code:
print(f"[卖出跳过] 指令{instruction_id}缺少股票代码")
return False
if not sell_shares or sell_shares <= 0:
print(f"[卖出跳过] 指令{instruction_id}股数无效: {sell_shares}股")
return False
iquant_code = convert_stock_code(orion_code)
if not iquant_code:
print(f"[卖出跳过] 指令{instruction_id}代码转换失败: {orion_code}")
return False
sell_shares = int(sell_shares / 100) * 100
if sell_shares <= 0:
print(f"[卖出跳过] 指令{instruction_id}股数无效: {sell_shares}股")
return False
market = get_market_from_code(orion_code)
formatted_price = format_price(sell_price, market)
print(f"\n[卖出执行] ID={instruction_id}")
print(f" 代码: {orion_code} -> {iquant_code}")
print(f" 价格: {formatted_price}, 股数: {sell_shares}")
try:
state.pending_orders[str(instruction_id)] = {
"instruction_id": instruction_id,
"type": "sell",
"stock_code": iquant_code,
"expected_shares": sell_shares,
"expected_price": formatted_price,
"order_time": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"status": "submitted",
"order_sys_id": None
}
remark = f"INSTR_{instruction_id}"
passorder(
24, 1101, TRADE_ACCOUNT, iquant_code, 11, formatted_price,
sell_shares, ORDER_CONFIG["strategy_name"], ORDER_CONFIG["quick_trade"],
remark, ContextInfo
)
print(f"[卖出提交] 指令{instruction_id}已提交，remark={remark}")
return True
except Exception as e:
print(f"[卖出异常] {str(e)}")
if str(instruction_id) in state.pending_orders:
del state.pending_orders[str(instruction_id)]
return False
def _execute_callback_with_retry(instruction_id, trade_type, real_price):
"""执行回写操作"""
task_key = f"{trade_type}_{instruction_id}"
if task_key in state.executed_instructions:
print(f"[回写] 指令{instruction_id}已执行过，跳过")
return False
if trade_type == "buy":
exec_data = {"id": instruction_id, "realbuyprice": real_price}
api_url = ORION_CONFIG["api_exec_buy"]
else:
exec_data = {"id": instruction_id, "realsellprice": real_price}
api_url = ORION_CONFIG["api_exec_sell"]
success, _ = _call_orion_api_post_form(api_url, exec_data)
if success:
state.executed_instructions.add(task_key)
print(f"[回写成功] {trade_type}指令{instruction_id}, 价格={real_price}")
return True
else:
print(f"[回写失败] {trade_type}指令{instruction_id}")
return False
def parse_remark(remark):
"""从remark中解析指令ID"""
if not remark:
return None
if remark.startswith("INSTR_"):
try:
instruction_id = remark.replace("INSTR_", "")
return int(instruction_id)
except ValueError:
return None
return None
def order_callback(ContextInfo, orderInfo):
"""委托状态变化回调"""
global state
print("\n" + "=" * 60)
print("[委托回调] 委托状态变化")
try:
order_sys_id = getattr(orderInfo, 'm_strOrderSysID', 'N/A')
stock_code = getattr(orderInfo, 'm_strInstrumentID', 'N/A')
order_status = getattr(orderInfo, 'm_nOrderStatus', 'N/A')
volume = getattr(orderInfo, 'm_nVolumeTotalOriginal', 0)
price = getattr(orderInfo, 'm_dLimitPrice', 0)
remark = getattr(orderInfo, 'm_strRemark', '')
# 匹配缓存中的订单
matched_order = None
direction_text = "未知"
for key, order_info in state.pending_orders.items():
cached_code = order_info.get("stock_code", "")
cached_code_clean = cached_code.split('.')[0]
if cached_code_clean == stock_code:
matched_order = order_info
break
if matched_order:
trade_type = matched_order.get("type")
direction_text = "买入" if trade_type == "buy" else "卖出"
print(f"[委托] 委托号: {order_sys_id}")
print(f"[委托] 股票: {stock_code}")
print(f"[委托] 方向: {direction_text}")
print(f"[委托] 价格: {price}, 数量: {volume}")
print(f"[委托] 状态码: {order_status}")
print(f"[委托] 备注: {remark}")
# 解析指令ID
instruction_id = parse_remark(remark)
if instruction_id is not None:
instr_key = str(instruction_id)
if instr_key in state.pending_orders:
state.pending_orders[instr_key]["order_sys_id"] = order_sys_id
print(f"[委托回调] [成功] 通过remark匹配到指令ID={instruction_id}")
print(f"[委托回调] 已绑定委托号 {order_sys_id}")
else:
print(f"[委托回调] [失败] 指令ID={instruction_id} 不在pending_orders中")
else:
print(f"[委托回调] [失败] 无法从remark解析指令ID: {remark}")
except Exception as e:
print(f"[委托回调异常] {str(e)}")
import traceback
traceback.print_exc()
print("=" * 60)
def fetch_and_execute_buy_orders(ContextInfo, session_type):
"""
获取并执行买入指令（增量获取）
用于卖出成功后重新获取新的买单
"""
global state
if not is_trading_time():
print(f"[{session_type}] 当前不在交易时段，跳过")
return
now = datetime.now()
today = now.strftime('%Y-%m-%d')
print(f"\n{'='*50}")
print(f"[{session_type}] 增量获取买单 - {now.strftime('%Y-%m-%d %H:%M:%S')}")
buy_data = {
"sdate": today,
"edate": today,
"cid": ORION_CONFIG["cid"],
"resultColumnIDs": ORION_CONFIG["result_column_ids"]
}
success, buy_instructions = _call_orion_api_post_form(ORION_CONFIG["api_get_buy"], buy_data, max_retries=2)
if success and buy_instructions and len(buy_instructions) > 0:
# 过滤出新的指令（未执行的）
new_instructions = []
for inst in buy_instructions:
inst_id = str(inst.get("ID"))
# 检查是否已执行过（在executed_instructions中）
if f"buy_{inst_id}" in state.executed_instructions:
continue
# 检查是否已在pending_orders中
if inst_id in state.pending_orders:
continue
new_instructions.append(inst)
if new_instructions:
print(f"[{session_type}] 获取到 {len(new_instructions)} 条新买入指令")
for inst in new_instructions:
original_price = inst.get("BuyPrice", 0)
market = get_market_from_code(inst.get("Code", ""))
formatted_price = format_price(original_price, market)
inst["BuyPrice"] = formatted_price
print(f" 新指令 {inst.get('ID')}: {original_price} -> {formatted_price}")
execute_buy_order(ContextInfo, inst)
else:
print(f"[{session_type}] 无新买入指令")
else:
print(f"[{session_type}] 获取买单失败或无指令")
def deal_callback(ContextInfo, dealInfo):
"""
成交回调 - 核心：卖出成功后触发增量买单获取
"""
global state
print("\n" + "=" * 60)
print("[成交回调] 收到成交信息")
try:
stock_code = getattr(dealInfo, 'm_strInstrumentID', 'N/A')
trade_price = getattr(dealInfo, 'm_dPrice', 0)
trade_volume = getattr(dealInfo, 'm_nVolume', 0)
direction = getattr(dealInfo, 'm_nDirection', 0)
order_sys_id = getattr(dealInfo, 'm_strOrderSysID', None)
trade_id = getattr(dealInfo, 'm_strTradeID', 'N/A')
remark = getattr(dealInfo, 'm_strRemark', '')
# 匹配缓存中的订单
matched_order = None
direction_text = "未知"
for key, order_info in state.pending_orders.items():
cached_code = order_info.get("stock_code", "")
cached_code_clean = cached_code.split('.')[0]
if cached_code_clean == stock_code:
matched_order = order_info
break
if matched_order:
trade_type = matched_order.get("type")
direction_text = "买入" if trade_type == "buy" else "卖出"
print(f"[成交] 委托号: {order_sys_id}")
print(f"[成交] 成交编号: {trade_id}")
print(f"[成交] 股票: {stock_code}")
print(f"[成交] 方向: {direction_text}")
print(f"[成交] 价格: {trade_price}, 数量: {trade_volume}")
print(f"[成交] 备注: {remark}")
# 通过remark解析指令ID
instruction_id = parse_remark(remark)
if instruction_id is not None:
instr_key = str(instruction_id)
if instr_key in state.pending_orders:
order_info = state.pending_orders[instr_key]
trade_type = order_info.get("type")
expected_price = order_info.get("expected_price", 0)
print(f"[成交] [成功] 通过remark匹配到指令ID={instruction_id}")
print(f"[成交] 订单类型: {trade_type}")
# 执行回写
callback_success = _execute_callback_with_retry(
instruction_id, trade_type, trade_price
)
# ========== 核心改进：卖出成功后触发增量买单获取 ==========
if callback_success and trade_type == "sell":
print(f"\n{'#'*50}")
print(f"[循环交易] 卖出指令{instruction_id}回写成功！")
print(f"[循环交易] 资金已释放，{REFRESH_BUY_INTERVAL}秒后获取新买单...")
print(f"{'#'*50}\n")
# 使用线程延迟执行，不阻塞主流程
def delayed_fetch_buy():
time.sleep(REFRESH_BUY_INTERVAL)
# 在handlebar中标记需要获取买单
state.pending_buy_fetch = True
threading.Thread(target=delayed_fetch_buy, daemon=True).start()
# 立即更新订单状态
order_info["status"] = "filled_sold"
order_info["actual_price"] = trade_price
order_info["actual_shares"] = trade_volume
# 从pending中移除
del state.pending_orders[instr_key]
print(f"[成交] 卖出订单已全部成交并回写，从待处理列表中移除")
elif callback_success and trade_type == "buy":
# 买入成功处理
order_info["status"] = "filled_bought"
order_info["actual_price"] = trade_price
order_info["actual_shares"] = trade_volume
expected_shares = order_info.get("expected_shares", 0)
if trade_volume >= expected_shares:
del state.pending_orders[instr_key]
print(f"[成交] 买入订单已全部成交，从待处理列表中移除")
else:
remaining = expected_shares - trade_volume
state.pending_orders[instr_key]["expected_shares"] = remaining
print(f"[成交] 买入部分成交，剩余数量: {remaining}")
else:
# 回写失败，保留订单待重试
print(f"[成交] 回写失败，保留订单待重试")
else:
print(f"[成交] [失败] 指令ID={instruction_id} 不在pending_orders中")
else:
# 降级：尝试用委托号匹配
print(f"[成交] 无法从remark解析指令ID，尝试委托号匹配...")
if order_sys_id and order_sys_id != 'N/A':
matched_key = None
for key, order_info in state.pending_orders.items():
if order_info.get("order_sys_id") == order_sys_id:
matched_key = key
break
if matched_key:
order_info = state.pending_orders[matched_key]
instruction_id = order_info.get("instruction_id")
trade_type = order_info.get("type")
print(f"[成交] 通过委托号匹配到指令ID={instruction_id}")
callback_success = _execute_callback_with_retry(
instruction_id, trade_type, trade_price
)
if callback_success and trade_type == "sell":
print(f"\n{'#'*50}")
print(f"[循环交易] 卖出指令{instruction_id}回写成功！")
print(f"[循环交易] 资金已释放，{REFRESH_BUY_INTERVAL}秒后获取新买单...")
print(f"{'#'*50}\n")
def delayed_fetch_buy_legacy():
time.sleep(REFRESH_BUY_INTERVAL)
state.pending_buy_fetch = True
threading.Thread(target=delayed_fetch_buy_legacy, daemon=True).start()
del state.pending_orders[matched_key]
else:
print(f"[成交] 回写失败，保留订单待重试")
else:
print(f"[成交] [失败] 未找到匹配订单")
else:
print(f"[成交] [失败] 无法匹配订单")
print(f"[成交] 成交明细: {stock_code} {'买入' if direction == 48 else '卖出'} {trade_volume}股 @ {trade_price}")
except Exception as e:
print(f"[成交回调异常] {str(e)}")
import traceback
traceback.print_exc()
print("=" * 60)
def process_morning_instructions(ContextInfo, session_type):
"""早盘处理 - 获取买入和卖出指令"""
now = datetime.now()
today = now.strftime('%Y-%m-%d')
print(f"\n{'='*50}")
print(f"[{session_type}] 开始获取指令 - {now.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"[{session_type}] 模式: 买入 + 卖出")
# 获取买入指令
buy_data = {
"sdate": today,
"edate": today,
"cid": ORION_CONFIG["cid"],
"resultColumnIDs": ORION_CONFIG["result_column_ids"]
}
success, buy_instructions = _call_orion_api_post_form(ORION_CONFIG["api_get_buy"], buy_data, max_retries=2)
if success and buy_instructions and len(buy_instructions) > 0:
print(f"[{session_type}] 买入: 获取到 {len(buy_instructions)} 条指令")
for inst in buy_instructions:
original_price = inst.get("BuyPrice", 0)
market = get_market_from_code(inst.get("Code", ""))
formatted_price = format_price(original_price, market)
inst["BuyPrice"] = formatted_price
print(f" 指令{inst.get('ID')}: {original_price} -> {formatted_price}")
execute_buy_order(ContextInfo, inst)
else:
print(f"[{session_type}] 买入: 无有效指令")
# 获取卖出指令
sell_data = {
"sdate": today,
"edate": today,
"cid": ORION_CONFIG["cid"],
"resultColumnIDs": ORION_CONFIG["result_column_ids"]
}
success, sell_instructions = _call_orion_api_post_form(ORION_CONFIG["api_get_sell"], sell_data, max_retries=2)
if success and sell_instructions and len(sell_instructions) > 0:
print(f"[{session_type}] 卖出: 获取到 {len(sell_instructions)} 条指令")
for inst in sell_instructions:
original_price = inst.get("BuyPrice", 0)
market = get_market_from_code(inst.get("Code", ""))
formatted_price = format_price(original_price, market)
inst["BuyPrice"] = formatted_price
print(f" 指令{inst.get('ID')}: {original_price} -> {formatted_price}")
execute_sell_order(ContextInfo, inst)
else:
print(f"[{session_type}] 卖出: 无有效指令")
print(f"[{session_type}] 完成")
print("=" * 50)
def process_close_instructions(ContextInfo, session_type):
"""收盘前处理 - 只获取卖出指令"""
global state
now = datetime.now()
today = now.strftime('%Y-%m-%d')
print(f"\n{'='*50}")
print(f"[{session_type}] 开始获取指令 - {now.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"[{session_type}] 模式: 仅卖出")
# 只获取卖出指令
sell_data = {
"sdate": today,
"edate": today,
"cid": ORION_CONFIG["cid"],
"resultColumnIDs": ORION_CONFIG["result_column_ids"]
}
success, sell_instructions = _call_orion_api_post_form(ORION_CONFIG["api_get_sell"], sell_data, max_retries=2)
if success and sell_instructions and len(sell_instructions) > 0:
print(f"[{session_type}] 卖出: 获取到 {len(sell_instructions)} 条指令")
for inst in sell_instructions:
original_price = inst.get("BuyPrice", 0)
market = get_market_from_code(inst.get("Code", ""))
formatted_price = format_price(original_price, market)
inst["BuyPrice"] = formatted_price
print(f" 指令{inst.get('ID')}: {original_price} -> {formatted_price}")
execute_sell_order(ContextInfo, inst)
else:
print(f"[{session_type}] 卖出: 无有效指令")
print(f"[{session_type}] 完成")
print("=" * 50)
# 收盘后清理残留订单
if len(state.pending_orders) > 0:
print(f"[清理] 发现 {len(state.pending_orders)} 个残留订单，正在清理...")
state.pending_orders.clear()
print(f"[清理] 已完成清理")
def check_and_execute(ContextInfo):
"""检查时间窗口并执行任务"""
global state
if state.is_executing:
return
# ========== 处理延迟的买单获取 ==========
if state.pending_buy_fetch:
state.pending_buy_fetch = False
state.is_executing = True
try:
fetch_and_execute_buy_orders(ContextInfo, "循环交易(卖出触发)")
finally:
state.is_executing = False
return
now = datetime.now()
today = now.strftime('%Y-%m-%d')
current_hour = now.hour
current_minute = now.minute
current_weekday = now.weekday()
# 周末不交易
if current_weekday >= 5 and not TEST_MODE:
return
# 测试模式
if TEST_MODE:
state.is_executing = True
try:
process_morning_instructions(ContextInfo, "测试模式")
finally:
state.is_executing = False
return
# 早盘时间窗口 - 买入 + 卖出
if (current_hour == MORNING_HOUR and
MORNING_MIN_START <= current_minute <= MORNING_MIN_END):
if state.last_morning_date != today:
print(f"\n[时间触发] 进入早盘时间窗口")
state.is_executing = True
try:
process_morning_instructions(ContextInfo, "早盘")
state.last_morning_date = today
finally:
state.is_executing = False
return
# 收盘前时间窗口 - 仅卖出
if (current_hour == CLOSE_HOUR and
CLOSE_MIN_START <= current_minute <= CLOSE_MIN_END):
if state.last_close_date != today:
print(f"\n[时间触发] 进入收盘前时间窗口")
state.is_executing = True
try:
process_close_instructions(ContextInfo, "收盘前")
state.last_close_date = today
finally:
state.is_executing = False
return
def handlebar(ContextInfo):
"""主循环 - 持续检查"""
global state
# 限制检查频率（每5秒检查一次）
current_time = time.time()
if current_time - state.last_check_time < 5:
return
state.last_check_time = current_time
check_and_execute(ContextInfo)
def position_callback(ContextInfo, positionInfo):
pass
def account_callback(ContextInfo, accountInfo):
pass第一步：获取策略代码
策略代码已在上方完整展示，包含以下核心功能：
- 定时以POST方式调用猎户座API获取买卖信号
- 解析API返回的JSON股票列表并提取关键信息
- 通过iQuant交易接口执行下单/撤单操作
- 成交后以POST方式回传交割指令到猎户座系统
- V8.0核心特性：卖单成功后立即提取新的买单信号，实现持续高效交易
第二步：在iQuant中创建策略
- 打开iQuant量化终端并登录
- 点击左侧菜单栏"策略管理" → "新建策略"
- 在弹出的对话框中：
- 策略类型：选择 Python策略
- 策略名称：输入"猎户座量化策略V8.0"
- 策略描述：可填写"猎户座AI量化交易策略 - 支持卖单后持续交易"
- 点击"确定"创建策略
- 将上方完整的Python代码复制粘贴到代码编辑器中
第三步：修改策略中的关键配置
在策略代码中找到以下配置项，必须替换为你自己的信息：
python
# ====== 猎户座API配置（必须修改）======
ORION_CONFIG = {
"appid": "你的AppID", # 替换为猎户座"账号配置"中的AppID
"appsecret": "你的AppSecret", # 替换为猎户座"安全中心"中的AppSecret
"cid": 1, # 1=A股, 2=港股, 3=美股
"result_column_ids": "4,5",
"api_get_buy": "http://www.deephunt.com.cn/api/getbuystock",
"api_get_sell": "http://www.deephunt.com.cn/api/getsellstock",
"api_exec_buy": "http://www.deephunt.com.cn/api/executeuserbuytrade",
"api_exec_sell": "http://www.deephunt.com.cn/api/executeuserselltrade",
}
TRADE_ACCOUNT = "xxxxxx" # 国信量化实盘账号第四步：配置策略运行参数
在iQuant策略编辑界面点击"运行设置"：
| 参数 | 设置建议 | 说明 |
|---|---|---|
| 运行周期 | 每分钟 | 确保及时获取新信号 |
| 运行时段 | 9:30-11:30, 13:00-15:00 | A股交易时段 |
| 自动重启 | 开启 | 策略异常退出后自动重启 |
| 日志级别 | 详细 | 便于问题排查 |
第五步：保存并编译策略
- 点击代码编辑器上方的"保存"按钮
- 点击"编译"按钮检查代码语法
- 如果编译报错，请检查：
- Python语法是否正确
- 所有括号是否匹配
- 缩进是否一致（建议使用4个空格）
重要说明：
- 首次运行前请确保已正确配置API认证信息
- 建议先在模拟交易模式下测试一周，确认无误后再切换实盘
6.4 开盘时段测试连通状态
重要：策略代码部署完成后，务必在开盘时段进行连通性测试，确保所有环节正常工作。
测试步骤：
选择测试时间：建议在交易日的上午9:35-9:45之间测试（此时交易已开始，API返回真实信号）
启动策略：在iQuant中点击"运行策略"按钮，观察启动日志
观察策略日志：确认以下关键信息：
- 策略成功初始化（显示"猎户座量化交易策略 V8.0 启动"）
- 成功调用API获取买卖信号
- 正确读取账户信息和持仓
手动验证API连通性（猎户座API使用 POST 方式）：
方式一：使用curl命令（推荐，适用于Linux/Mac）
bashcurl -X POST "http://www.deephunt.com.cn/api/getbuystock" \ -H "Content-Type: application/x-www-form-urlencoded" \ -d "appid=你的AppID&appsecret=你的AppSecret&cid=1&sdate=2024-01-01&edate=2024-12-31"方式二：使用PowerShell（适用于Windows服务器）
powershell$body = @{ appid = "你的AppID" appsecret = "你的AppSecret" cid = "1" sdate = "2024-01-01" edate = "2024-12-31" } Invoke-WebRequest -Uri "http://www.deephunt.com.cn/api/getbuystock" ` -Method POST -ContentType "application/x-www-form-urlencoded" -Body $body | Select-Object -ExpandProperty Content方式三：使用Python脚本验证
pythonimport requests # 配置你的API信息 APP_ID = "你的AppID" APP_SECRET = "你的AppSecret" url = "http://www.deephunt.com.cn/api/getbuystock" data = { "appid": APP_ID, "appsecret": APP_SECRET, "cid": "1", # A股市场 "sdate": "2024-01-01", "edate": "2024-12-31" } try: response = requests.post(url, data=data, headers={"Content-Type": "application/x-www-form-urlencoded"}, timeout=30) print(f"状态码: {response.status_code}") result = response.json() print(f"响应代码: {result.get('code')}") print(f"响应消息: {result.get('message')}") if result.get('code') == 1: print(f"信号数量: {len(result.get('data', []))}") print("✓ API连通测试成功！") else: print("✗ API返回错误") except Exception as e: print(f"✗ 请求失败: {str(e)}")验证标准：如果返回的JSON中
code字段为1，说明API连通正常。检查点清单：
检查项 预期结果 异常处理步骤 策略启动 无报错，显示启动日志 检查Python语法、缩进是否正确、配置项是否完整 API调用（POST） 返回JSON格式，code=1 检查IP白名单、AppID/AppSecret是否正确、API地址是否正确 获取买卖信号 返回信号列表（可能为空） 检查日期参数格式、市场ID是否正确、是否在交易时段调用 账户信息读取 显示正确的资金和持仓数据 检查iQuant是否已登录、账户是否有交易权限 下单接口测试 模拟下单成功（无报错） 检查股票代码格式、价格是否合理、数量是否为100的整数倍
提示：测试阶段建议在账号配置中选择"模拟交易"模式，运行至少1周确认策略稳定后，再切换为"实盘交易"。
注意：首次运行时，如果出现"IP不在白名单"错误，请登录猎户座安全中心将服务器公网IP添加到白名单中。
6.5 配置自动运行与自动关闭
为什么需要配置自动运行与自动关闭？
国信证券iQuant量化终端存在一个已知问题：软件会在运行一段时间后自动关闭，且关闭后无法自动重启。这会导致量化策略中断，错过交易机会。因此，必须在Windows服务器上配置自动化任务，确保：
- 开盘前自动启动iQuant：确保交易时段开始前软件已就绪
- 收盘后自动关闭iQuant：正常结束当天交易
- 运行中异常重启：如果iQuant意外关闭，自动重新启动
- 服务器定期重启：保证系统长期稳定运行
参考文档：点击下载 3-window服务器上自动运行与关闭量化终端.docx（在新窗口打开后可下载）
📄 Windows服务器自动运行与关闭量化终端（完整内容）
================================================================
Windows服务器上自动运行与关闭量化终端
================================================================
一、概述
--------
量化交易需要无人值守运行，本文档介绍如何在Windows服务器上配置：
1. 开盘前自动启动iQuant和策略
2. 收盘后自动关闭iQuant
3. 服务器定时重启（保证长期稳定运行）
4. 异常情况监控与告警
二、方案一：Windows任务计划程序（推荐新手）
------------------------------------------
### 2.1 配置开盘前自动启动iQuant
步骤：
1. 按 Win + R，输入 taskschd.msc，回车
2. 点击右侧"创建基本任务"
3. 名称：输入"开盘启动iQuant"
4. 触发器：选择"每天"
5. 时间：设置为 9:25（开盘前5分钟）
6. 操作：选择"启动程序"
7. 程序路径：浏览选择iQuant的exe文件
例如：C:\iQuant\iQuant.exe
8. 点击"完成"
### 2.2 配置iQuant内策略自动启动
在iQuant中设置策略自动启动：
1. 打开iQuant → 策略管理
2. 找到你的策略，右键 → "属性"
3. 勾选"启动iQuant时自动运行此策略"
4. 勾选"策略异常停止后自动重启"
5. 保存设置
### 2.3 配置收盘后自动关闭iQuant
步骤：
1. 在任务计划程序中创建新任务
2. 名称：输入"收盘关闭iQuant"
3. 触发器：选择"每天"
4. 时间：设置为 15:05（收盘后5分钟）
5. 操作：选择"启动程序"
6. 程序或脚本输入：taskkill
7. 参数输入：/F /IM iQuant.exe
8. 点击"完成"
### 2.4 任务计划程序高级设置
对于每个任务，建议在"属性"中进行以下设置：
常规选项卡：
- 勾选"不管用户是否登录都要运行"
- 勾选"使用最高权限运行"
条件选项卡：
- 取消勾选"只有在计算机使用交流电源时才启动"
- 取消勾选"如果计算机切换到电池电源则停止"
设置选项卡：
- 勾选"如果任务失败，按以下频率重新启动"
- 设置重试间隔：1分钟
- 设置重试次数：3次
三、方案二：批处理脚本（更灵活）
--------------------------------
### 3.1 创建启动脚本
创建文件 C:\scripts\start_iquant.bat：
@echo off
chcp 65001 >nul
title 猎户座量化交易启动脚本
echo ========================================
echo 猎户座量化交易系统 - 启动中...
echo 时间: %date% %time%
echo ========================================
REM 1. 检查iQuant是否已在运行
tasklist /FI "IMAGENAME eq iQuant.exe" 2>NUL | find /I /N "iQuant.exe" >NUL
if "%ERRORLEVEL%"=="0" (
echo [警告] iQuant已在运行中，先关闭旧进程...
taskkill /F /IM iQuant.exe
timeout /t 5 /nobreak
)
REM 2. 启动iQuant
echo [启动] 正在启动iQuant...
start "" "C:\iQuant\iQuant.exe"
REM 3. 等待iQuant完全启动
echo [等待] 等待iQuant初始化（30秒）...
timeout /t 30 /nobreak
echo ========================================
echo 量化交易系统启动完成！
echo 时间: %date% %time%
echo ========================================
### 3.2 创建关闭脚本
创建文件 C:\scripts\stop_iquant.bat：
@echo off
chcp 65001 >nul
title 猎户座量化交易关闭脚本
echo ========================================
echo 猎户座量化交易系统 - 关闭中...
echo ========================================
REM 优雅关闭iQuant
taskkill /IM iQuant.exe
timeout /t 10 /nobreak
REM 强制关闭（如果进程仍在运行）
tasklist /FI "IMAGENAME eq iQuant.exe" 2>NUL | find /I /N "iQuant.exe" >NUL
if "%ERRORLEVEL%"=="0" (
echo [强制] iQuant未正常关闭，强制结束进程...
taskkill /F /IM iQuant.exe
)
echo ========================================
echo 量化交易系统已关闭！
echo ========================================
四、方案三：全天候运行 + 时段内交易（最稳定）
--------------------------------------------
保持iQuant全天运行，策略代码内部判断交易时段：
```python
import datetime
def is_trading_time():
now = datetime.datetime.now()
morning_start = now.replace(hour=9, minute=30, second=0)
morning_end = now.replace(hour=11, minute=30, second=0)
afternoon_start = now.replace(hour=13, minute=0, second=0)
afternoon_end = now.replace(hour=15, minute=0, second=0)
return (morning_start <= now <= morning_end) or \
(afternoon_start <= now <= afternoon_end)优点：
- 不需要频繁启动关闭iQuant
- 策略在非交易时段自动休眠
- 更稳定，减少因启动关闭引发的问题
五、服务器定时重启（保证长期稳定）
创建任务计划：
- 名称：每周日凌晨重启服务器
- 触发器：每周日，凌晨 3:00
- 操作：启动程序
- 程序：shutdown
- 参数：/r /f /t 0
六、异常监控方案
创建监控脚本 C:\scripts\check_iquant.bat：
@echo off tasklist /FI "IMAGENAME eq iQuant.exe" 2>NUL | find /I /N "iQuant.exe" >NUL if "%ERRORLEVEL%"=="0" ( echo iQuant运行正常 ) else ( echo [异常] iQuant未运行，正在重新启动... call C:\scripts\start_iquant.bat )
在任务计划程序中设置此脚本每5分钟运行一次。
================================================================
**方案一：使用Windows任务计划程序（推荐新手）**
**配置自动启动iQuant（开盘前）**：
1. 按 `Win + R`，输入 `taskschd.msc`，回车
2. 点击"创建基本任务"，名称输入"开盘启动iQuant"
3. 触发器选择"每天"，时间设置为 **9:25**
4. 操作选择"启动程序"，浏览选择iQuant的exe路径
5. 点击"完成"
**配置自动启动策略**：在iQuant中勾选"策略随软件启动"选项。
**配置自动关闭iQuant（收盘后）**：
1. 创建新任务，名称"收盘关闭iQuant"
2. 触发器：每天 **15:05**
3. 操作：启动程序 `taskkill`，参数 `/F /IM iQuant.exe`
**方案二：使用批处理脚本（更灵活）**
```batch
@echo off
start "" "C:\iQuant\iQuant.exe"
timeout /t 30 /nobreak
echo 量化交易系统已启动 -- %date% %time%方案三：全天候运行 + 时段内交易（最稳定）
iQuant保持全天运行，策略代码内部判断交易时段（已在6.3节的策略代码中内置 is_trading_time() 函数）。
七、完整部署检查清单
权限检查
- [ ] 猎户座量化交易权限已开通 → 量化交易操作指南
- [ ] 券商量化交易权限已开通（收到券商确认通知）
- [ ] iQuant账号已获取，能正常登录
服务器检查（阿里云）
- [ ] 阿里云ECS服务器已购买并正常运行
- [ ] 远程桌面可正常登录
- [ ] 服务器已设置永不休眠
- [ ] 服务器已配置自动登录（netplwiz）
软件检查
- [ ] iQuant量化终端已安装
- [ ] iQuant已使用实盘账号登录成功
- [ ] 能在iQuant中看到账户信息和持仓
策略检查
- [ ] 策略代码已部署到iQuant
- [ ] AppID和AppSecret已正确配置
- [ ] 策略代码无语法错误
连通性检查
- [ ] 服务器公网IP已加入猎户座IP白名单 → 安全中心
- [ ] API接口POST方式返回JSON，
code为1 - [ ] 开盘时段策略能正常获取买卖信号
自动化检查
- [ ] 自动启动任务已配置（开盘前9:25）
- [ ] 自动关闭任务已配置（收盘后15:05）
- [ ] 或策略代码中包含交易时段判断
安全提醒
八、常见问题解答
权限相关
Q：猎户座申请开通量化交易后多久能审核通过？ A：通常即时生效，如果长时间未生效，请联系客服。
Q：券商开通量化交易需要什么条件？ A：一般需要开立证券账户满3个月以上，无不良记录，具体以券商要求为准。详见上方"申请报备实例"章节。
Q：可以用模拟交易吗？需要申请权限吗？ A：猎户座的模拟交易无需连接券商，在账号配置中选择"模拟交易"模式即可。
服务器相关（阿里云）
Q：能用家里的电脑代替云服务器吗？ A：不推荐。家用电脑可能因断电、断网、休眠等原因导致交易中断。本指南以阿里云ECS为例演示部署流程，你也可以选择腾讯云、华为云等其他云服务商。云服务器有专业机房保障，稳定性更高。
Q：阿里云服务器需要一直开着吗？ A：如果选择方案三（全天候运行），需要24小时开机。如果选择方案一（定时开关），可以在非交易时段关机节省费用。按量付费模式下，关机不收费。
Q：Mac用户能用吗？ A：iQuant仅支持Windows。Mac用户推荐使用阿里云Windows服务器，通过Microsoft Remote Desktop连接。
Q：阿里云服务器如何选择地域？ A：选择离券商服务器近的地域，如华东1（杭州）、华东2（上海）、华南1（深圳）。
Q：阿里云服务器可以同时运行多个策略吗？ A：可以。4核8G服务器最多运行2-3个策略。
策略部署相关
Q：策略代码在哪里获取？ A：策略代码已在上方第六章完整展示，可直接复制使用。也可下载 .docx 参考文档。
Q：策略代码需要懂编程吗？ A：基本的配置修改（如填写AppID、AppSecret）不需要编程知识。
Q：策略运行后没有报错，但也没有交易，正常吗？ A：正常。如果当天没有符合你交易技能Skills的买卖信号，策略就不会产生交易。
Q：API调用返回错误怎么办？ A：常见错误及解决方法：
- "权限不足"：检查AppID和AppSecret是否正确
- "IP不在白名单"：检查服务器IP是否已加入安全中心的白名单
- "请求超时"：检查服务器网络连接，尝试ping deephunt.com.cn
- "参数错误"：检查API请求参数格式，注意要用POST方式提交
连通性相关
Q：为什么API必须用POST方式？ A：猎户座API使用POST方式提交参数，Content-Type为 application/x-www-form-urlencoded。GET方式不被支持。详见6.4节的手动验证方法。
Q：为什么我的IP白名单配置后还是无法访问API？ A：可能原因：服务器出口IP与查询到的IP不一致；白名单未保存生效；阿里云安全组限制了出站流量。用 curl ifconfig.me 确认当前出口IP。
Q：API调用频率有限制吗？ A：建议每分钟不超过10次请求。策略运行间隔1分钟完全在限制范围内。
自动运行相关
Q：Windows任务计划程序设置后没生效怎么办？ A：检查任务是否启用、触发器时间、程序路径。在"条件"中取消勾选"只有在计算机使用交流电源时才启动"，在"设置"中勾选"如果任务失败则重新启动"。
Q：服务器重启后策略会自动运行吗？ A：如果配置了自动登录 + Windows任务计划程序，会自动恢复。建议定期检查。
Q：iQuant运行久了会卡顿或崩溃吗？ A：有可能。建议每周重启一次服务器，监控CPU和内存使用率（阿里云控制台提供监控面板）。
安全相关
Q：AppSecret泄露了怎么办？ A：立即登录猎户座 → 安全中心 → 重新生成AppSecret。旧密钥立即失效。
Q：如何防止策略异常交易？ A：在账号配置中设置合理的资金利用率（如50%）、持仓上限（如5只）、止盈止损。先在模拟模式运行至少1-2周。
Q：如果策略出现bug导致错误下单怎么办？ A：立即在iQuant中停止策略，手动撤销错误订单。标准策略代码已内置最大单日交易次数限制。
九、附录：关键信息速查
猎户座平台信息
| 项目 | 信息 |
|---|---|
| 平台地址 | http://www.deephunt.com.cn |
| API基础地址 | http://www.deephunt.com.cn/api |
| 获取买入股票API | POST /api/getbuystock |
| 获取卖出股票API | POST /api/getsellstock |
| 获取买卖信号API | POST /api/getstocksignal |
| 同步买单交割API | POST /api/executeuserbuytrade |
| 同步卖单交割API | POST /api/executeuserselltrade |
| 详细API文档 | API接口 |
国信证券联系信息
| 项目 | 信息 |
|---|---|
| 客服电话 | 95536 |
| 官网 | https://www.guosen.com.cn |
| iQuant下载 | 联系客户经理获取 |
推荐云服务器配置（阿里云）
| 配置项 | 推荐选择 | 参考价格 |
|---|---|---|
| 产品 | 云服务器ECS | - |
| 实例规格 | 4核8GB（ecs.c6.xlarge） | ~150-200元/月 |
| 系统盘 | SSD云盘 80GB | 包含在套餐内 |
| 带宽 | 固定带宽 3-5Mbps | 包含在套餐内 |
| 地域 | 华东1/2、华南1 | 无额外费用 |
参考文档下载
| 文档 | 下载链接 |
|---|---|
| iQuant量化终端部署核心流程 | 点击下载 |
| iQuant量化交易标准代码 V8.0 | 点击下载 |
| Windows服务器自动运行与关闭 | 点击下载 |
| 量化实盘权限申请报备实例 | 点击下载 |
猎户座相关文档
| 文档 | 说明 |
|---|---|
| 量化交易操作指南 | 量化交易功能完整介绍 |
| 账号配置 | 交易参数和资金配置说明 |
| 安全中心 | IP白名单和AppSecret配置 |
| 交易技能Skills | 自定义交易策略技能 |
| API接口 | 完整的API接口文档 |
| 历史收益 | 交易绩效分析 |
*本指南由猎户座量化交易团队编写，旨在帮助用户快速完成量化交易的全流程部署。如有疑问，欢迎联系客服获取支持。祝你在AI量化交易的道路上收获满满�