xingshu 1.0.2

专为中文母语开发者打造的量化交易研发框架。

星枢量化框架

初衷：刚开始是想单纯想写一个策略，对比多个框架后选择了vnpy，他的文档很详细，但我入门还是用了较长的时间去阅读文档和看他的源码，后来考虑到我不需要使用图形化界面和想以母语中文对我后续策略开发和策略维护更友好的想法，决定以vnpy中的examples/no_ui/run.py为基础，进行全面中文重构。

星枢（xingshu）—— 专为中文母语开发者打造的量化交易研发框架，致力于提供更友好的开发体验。作为本土化策略研发解决方案，它不仅深度契合中文开发者的思维习惯，更提供从策略设计到高效落地的全流程支持，助力开发者聚焦核心交易逻辑的构建与优化。

📖 项目背景

为满足中文开发者对策略代码可读性和可维护性的需求，本项目以[vnpy-3.9.4](vnpy/vnpy at 3.9.4)为基础，全面重构：

• 全中文命名体系：包/模块/类/变量100%中文标识
• 深度语义对应：保留原框架设计理念，建立精确中英术语映射
• 无缝兼容升级：在保持功能一致性的基础上提升中文开发体验

🚀 核心优势

中文友好性

• 完全消除英文术语障碍
• API设计符合中文编程思维习惯
• 文档字符串及注释全中文化

架构延续性

• 完整保留vnpy的优秀架构设计
• 事件驱动引擎、主引擎等核心机制保持不变
• 渐进式重构保证功能稳定性

开发便捷性

• 更直观的代码导航
• 降低新人学习成本
• 提升长期维护效率

开发进度看板

模块	进度	预计完成时间
事件引擎	✅ 100%	已发布
主引擎	✅ 100%	已发布
CTA自动交易模块	✅ 100%	已发布
CTP接口	✅ 100%	已发布

安装

pip install xingshu

使用

在任意目录下创建run.py，写入示例代码：

# 运行单个策略
# 运行所有策略的示例代码在本项目github下：运行示例 文件夹

import multiprocessing
import sys
from time import sleep
from datetime import datetime, time
from logging import INFO

from xingshu.包_事件引擎 import 类_事件引擎
from xingshu.包_交易核心.模块_设置 import 全局设置
from xingshu.包_交易核心.模块_主引擎 import 类_主引擎
from xingshu_ctp import 类_CTP网关

from xingshu-cta-jymokuai import 类_CTA策略应用
from xingshu-cta-jymokuai.模块_基础 import 事件类型_CTA日志


全局设置["日志.启用"] = True
全局设置["日志.级别"] = INFO
全局设置["日志.控制台"] = True

ctp_设置 = {
    "用户名": "",
    "密码": "",
    "经纪商代码": "",
    "交易服务器": "",
    "行情服务器": "",
    "产品名称": "",
    "授权编码": "",
    "产品信息": ""
}

# 策略配置列表（可扩展添加多个策略）
策略配置列表 = [
    {
        "策略类名": "示例策略_基础教程",
        "策略名称": "示例策略",
        "合约标识": "c2505.大商所",
        "参数设置": {
            "初始手数": 1
        }
    }
]

# 中国期货市场交易时段（日盘/夜盘）
日盘开始时间 = time(8, 45)
日盘结束时间 = time(15, 0)

夜盘开始时间 = time(20, 45)
夜盘结束时间 = time(2, 45)


def 检查交易时段():
    """检查是否在交易时段内"""
    当前时间 = datetime.now().time()

    是否交易时段 = False
    if (
            (当前时间 >= 日盘开始时间 and 当前时间 <= 日盘结束时间)
            or (当前时间 >= 夜盘开始时间)
            or (当前时间 <= 夜盘结束时间)
    ):
        是否交易时段 = True

    return 是否交易时段


def 运行子进程():
    """在子进程中运行的业务逻辑"""
    全局设置["日志.文件"] = True

    # 创建事件引擎
    事件引擎 = 类_事件引擎()

    # 创建主引擎
    主引擎 = 类_主引擎(事件引擎)
    主引擎.添加网关(类_CTP网关)

    # 添加CTA应用
    CTA引擎 = 主引擎.添加应用(类_CTA策略应用)
    主引擎.记录日志("主引擎创建成功")

    # 注册日志事件监听
    日志引擎 = 主引擎.获取引擎("日志")
    事件引擎.注册类型处理器(事件类型_CTA日志, 日志引擎.处理日志事件)
    主引擎.记录日志("注册日志事件监听")

    # 连接CTP接口
    主引擎.连接网关(ctp_设置, "CTP")
    主引擎.记录日志("连接CTP接口")

    # 等待10秒确保连接成功
    sleep(10)

    # 初始化CTA引擎
    CTA引擎.初始化引擎()
    主引擎.记录日志("CTA策略引擎初始化完成")

    # 循环处理所有策略配置
    for 策略配置 in 策略配置列表:
        策略名称 = 策略配置["策略名称"]

        if 策略名称 not in CTA引擎.运行策略字典:
            主引擎.记录日志(f"创建策略：{策略名称}")
            CTA引擎.添加策略(
                类名=策略配置["策略类名"],
                策略名称=策略名称,
                合约标识=策略配置["合约标识"],
                配置字典=策略配置["参数设置"]
            )
            # 初始化策略
            CTA引擎.初始化策略(策略名称)
            sleep(5)
            CTA引擎.启动策略(策略名称)
        else:
            主引擎.记录日志(f"更新策略参数：{策略名称}")
            CTA引擎.更新策略配置(策略名称=策略名称,配置字典=策略配置["参数设置"])
            # 初始化策略
            CTA引擎.初始化策略(策略名称)
            sleep(5)
            CTA引擎.启动策略(策略名称)

    # 持续运行检查
    while True:
        sleep(10)

        # 检查交易时段
        是否交易时段 = 检查交易时段()
        if not 是否交易时段:
            print("关闭子进程")
            主引擎.关闭()
            sys.exit(0)


def 运行父进程():
    """在父进程中运行的守护逻辑"""
    print("启动CTA策略守护父进程")

    子进程 = None

    while True:
        # 持续检查交易时段
        是否交易时段 = 检查交易时段()

        # 交易时段启动子进程
        if 是否交易时段 and 子进程 is None:
            print("启动子进程")
            子进程 = multiprocessing.Process(target=运行子进程)
            子进程.start()
            print("子进程启动成功")

        # 非交易时段关闭子进程
        if not 是否交易时段 and 子进程 is not None:
            if not 子进程.is_alive():
                子进程 = None
                print("子进程关闭成功")

        sleep(5)


if __name__ == "__main__":
    运行父进程()

策略示例

策略示例代码通常保存在： xingshu_ctastrategy\策略文件夹 中。

如果您自己编写了策略文件，可以选择将它们放置在该目录下，也可以在用户目录中创建一个新的文件夹来存放。对于 Windows 用户来说，自定义的策略文件夹路径通常为：

C:\Users\xxx\策略文件夹（其中 xxx 为您的用户名）。

🤝 贡献指南

欢迎提交中文命名的：

• 📝 文档翻译
• 🐛 Bug修复
• 🎯 功能增强