惯性聚合 高效追踪和阅读你感兴趣的博客、新闻、科技资讯
阅读原文 在惯性聚合中打开

推荐订阅源

博客园 - Franky
Microsoft Azure Blog
Microsoft Azure Blog
阮一峰的网络日志
阮一峰的网络日志
宝玉的分享
宝玉的分享
量子位
N
Netflix TechBlog - Medium
M
MIT News - Artificial intelligence
GbyAI
GbyAI
Apple Machine Learning Research
Apple Machine Learning Research
博客园_首页
博客园 - 叶小钗
Cyber Security Advisories - MS-ISAC
Cyber Security Advisories - MS-ISAC
酷 壳 – CoolShell
酷 壳 – CoolShell
T
Tailwind CSS Blog
Y
Y Combinator Blog
L
LangChain Blog
The Cloudflare Blog
T
The Blog of Author Tim Ferriss
U
Unit 42
Martin Fowler
Martin Fowler
aimingoo的专栏
aimingoo的专栏
G
Google Developers Blog
freeCodeCamp Programming Tutorials: Python, JavaScript, Git & More
月光博客
月光博客

博客园 - lsgxeva

SD 卡 / T F卡(Micro SD)检测工具解析:如何用硬件工具确保存储介质可靠性? 不用#ifdef写跨平台代码 Windbg常用调试命令 全频段阻塞 Practical Clean Architecture in Typescript, Rust & Python 摄像头配置 ClickToRunSvc logiccircuit help Python 模块 NumPy 影像处理 AI协作 Python 函数和类 Python 入门 VisionLink-RT VisionLink-AI-Glasse RK3588 firefly net n2n config OpenWrt 集成 openNDS 进行 强制门户(Captive Portal) 认证 OpenWrt 集成 easycwmpd 成为 TR-069协议中的设备端(CPE) eh.h acu100 bridge vlan config 卫星天线(KA 0.85m 4W_BUC)使用 SatBox331-20 Modem 的性能情况 OpenSpec OPSX 完整指南 Claude Skill Creator 2.0 完整上手攻略 Auto-Memory + CLAUDE.md Conductor 完整上手攻略 GitNexus 完整上手攻略 code-review-graph 完整上手攻略 Claude Code Hooks 完整开发者指南 Openwrt switch vlan配置 llm-course Claude Code 入门教程
第一个案例策略
lsgxeva · 2026-09-14 · via 博客园 - lsgxeva

第一个案例策略

来源 https://zhuanlan.zhihu.com/p/1960344138928074814

官方教程

首页 - PythonGO

策略

每个策略文件(*.py)都是一个策略,我们要使用这个策略,首先需要加载它,可以理解为 Python 中的import操作,而实例是通过策略创建出来的,每个策略可以创建 N 个实例,可以理解为 Python 中的「实例化」操作。

第一次双击可以加载策略,第二次双击可以创建实例

当你修改了策略文件的代码,是需要重新加载该策略的,否则修改后的代码不会生效。点击「策略管理」按钮,在策略管理窗口中「单击」准备重新加载的策略,点击「重载」按钮,即可完成重新加载。

实例

创建实例以后,点击左下角放大镜,修改参数。

修改实例的参数

.

参数调整

当我们创建好实例后,也正常运行了,运行之后有点击暂停实例,这个时候实例的参数信息会被保存到一个JSON文件,等下一次我们想再加载这个实例的时候,就可以省略加载策略的步骤,直接加载实例了。

运行实例

等我们完成了前面所有操作,那就可以运行该策略实例了,在 Python 代码中,我们一般是运行一个函数,比如 main(),或者任何我们想调用的函数;但是在无限易中,实际上是一系列的启动流程,我们可以把这个启动流程理解为初始化,无限易会去调用自带的回调函数来完成初始化,也可以理解为 PythonGO 的生命周期,在这个生命周期中,有「创建 - 初始化 - 启动 - 暂停」等一系列过程。那我们简单理解就是:无限易会按照顺序调用策略代码里的一些函数,来实现策略的完全初始化。

在我们点击运行策略按钮后,其实无限易只是简单的执行了策略代码中的 on_start 回调函数,那为什么只是执行一个函数,策略却会一直运行下去呢?

on_start案例代码

其实这是因为有行情数据来驱动,我们在 on_start 中会去订阅我们设置的合约的行情数据(订阅就是告诉无限易我需要哪个合约的数据,你给我推送过来),然后无限易会把行情数据通过 on_tick 回调函数推送过来,直到我们取消订阅该合约。有了源源不断的数据,我们的程序就能一直跑下去了,我们可以利用这些行情数据来合成 K 线,计算指标,超价报单或者任何你能想到的操作。

on_tick案例

定义参数映射

参数映射,是指可以通过无限易 PythonGO 窗口参数栏对策略的参数进行修改的映射操作

现阶段,我们的参数映射都是通过 pydantic 这个强大的数据验证库,它可以根据我们定义的类型注释严格控制数据类型,在我们写代码的过程中就可以感受到该库的强大之处。

我们通过一个「参数映射模型」即可完成映射

from typing import Literal
 
# 从 base 库中导入定义参数和状态映射模型必须的三个方法
from pythongo.base import BaseParams, BaseState, Field
 
 
class Params(BaseParams):
    """参数映射模型"""
    exchange: str = Field(default="", title="交易所代码")
    instrument_id: str = Field(default="", title="合约代码")
    order_price: int | float = Field(default=0, title="报单价格")
    order_volume: int = Field(default=1, title="报单手数")
    order_direction: Literal["buy", "sell"] = Field(default="buy", title="报单方向")

上面的代码表示:

我们自己的参数映射模型叫做 Params,继承自 BaseParams

我们想从无限易 PythonGO 窗口参数栏传入定义的这 5 个值(或者说修改 5 个值)

类型注释在参数映射模型中是一定要写的,所有的值类型都会按照定义的类型注释被解析和自动转换,其中 Literal 的作用是表示这个变量只允许使用定义好的字面值,当前代码表示 order_direction 只允许使用 buy 或者 sell 这两个字符串

Field 用来自定义元数据并将其添加到参数映射模型的字段中

default(必填)定义这个参数默认值,相当于 self.exchange = ""

title(必填)定义这个参数的中文名,会在无限易 PythonGO 窗口显示

效果如图所示:

上述代码结果

定义状态映射

状态映射,是指可以通过无限易 PythonGO 窗口状态栏对策略的参数或变量进行实时查看的映射操作(如没该需求可以不定义)

定义方法和参数映射是一样的,通过一个「状态映射模型」即可完成映射

# 从 base 库中导入定义参数和状态映射模型必须的三个方法
from pythongo.base import BaseParams, BaseState, Field
 
 
class State(BaseState):
    """状态映射模型"""
    order_id: int | None = Field(default=None, title="报单编号")

上面的代码表示:

  • 我们自己的状态映射模型叫做 State,继承自 BaseState
  • 我们想从无限易 PythonGO 窗口状态栏查看报单编号(order_id)的值(注意:这个报单编号需要我们自己赋值,和报单成功后的返回的报单编号没有主动映射关系,仅仅是同名,你也可以定义为任意名称)
  • 其他和参数映射说明一样

效果如图所示:

定义策略类

当我们把映射模型定义好之后,就要开始写策略类了

根据规范 策略规范 - 导入父类 导入我们需要的父类基础策略模版 BaseStrategy,然后基于我们之前学过的知识 运行原理 - 加载策略,我们知道了「策略类名」要和「策略文件名」保持一致:

# 从 base 库中导入基础策略模版 BaseStrategy 
from pythongo.base import BaseParams, BaseState, BaseStrategy, Field
 
 
class DemoTest(BaseStrategy):
    """我的第一个策略"""

根据 PythonGO 代码规范,我们还需要给策略写一个函数文档字符串,来告诉无限易这个策略的描述

效果如图所示:

实例化映射模型

上面定义好了两个映射模型,也定义好了策略类,现在我们把映射模型实例化,作为策略的实例属性,这样才可以在策略中获取到定义的映射数据

from typing import Literal
 
from pythongo.base import BaseParams, BaseState, BaseStrategy, Field
 
 
class Params(BaseParams):
    """参数映射模型"""
    exchange: str = Field(default="", title="交易所代码")
    instrument_id: str = Field(default="", title="合约代码")
    order_price: int | float = Field(default=0, title="报单价格")
    order_volume: int = Field(default=1, title="报单手数")
    order_direction: Literal["buy", "sell"] = Field(default="buy", title="报单方向")
 
 
class State(BaseState):
    """状态映射模型"""
    order_id: int | None = Field(default=None, title="报单编号")
 
 
class DemoTest(BaseStrategy):
    """我的第一个策略"""
    def __init__(self) -> None:
        super().__init__()
        self.params_map = Params()
        self.state_map = State()

定义回调

在 PythonGO 的父类基础策略模版中,以 on 开头的函数都是回调函数

在 运行原理 中,我们大致的了解了回调函数,那在当前的示例中我们只需要用到三个回调函数,根据 策略规范 - 回调函数 按顺序定义如下:

on_start()

查看启动策略回调定义

我们的策略的逻辑是一旦运行就按照无限易 PythonGO 窗口的参数下单,那下单函数就应该写在本回调中

我们使用 报单函数 - send_order 来进行报单操作,对应的报单数据我们从参数 self.params_map 中获取,得到的报单编号,我们赋值给状态映射模型中的 order_id,然后需要调用 update_status_bar 函数才可以刷新状态栏的值:

def on_start(self) -> None:
    super().on_start()
 
    self.state_map.order_id = self.send_order(
        exchange=self.params_map.exchange,
        instrument_id=self.params_map.instrument_id,
        volume=self.params_map.order_volume,
        price=self.params_map.order_price,
        order_direction=self.params_map.order_direction
    )

    self.update_status_bar() #update_status_bar函数定义:更新无限易 PythonGO 窗口状态栏显示值

on_order()

在报单变化回调中,我们只需要输出看看报单后回调收到的报单数据(OrderData)信息。要注意的是,参数order是一个实例对象,不是字典对象,所以他的取值是用.取值的,比如order.exchange,但是我们直接输出order,显示的内容确是字典,这是因为我们对这个数据对象做了处理,如果没做处理的话,看到的是内存地址<order.OrderData at 0x96b06c0>

from pythongo.classdef import OrderData
 
def on_order(self, order: OrderData) -> None:
    super().on_order(order)
    self.output("报单信息:", order)

完整代码

from typing import Literal  # 导入Literal类型,用于定义变量只能取特定值(如"buy"或"sell")

from pythongo.base import BaseParams, BaseState, BaseStrategy, Field  # 从pythongo基础模块导入核心类
# BaseParams:参数映射基类(用于定义策略参数)
# BaseState:状态映射基类(用于定义策略状态)
# BaseStrategy:策略基类(所有策略必须继承)
# Field:用于定义参数的元数据(如默认值、中文标题等)

from pythongo.classdef import OrderData  # 从pythongo类定义模块导入OrderData类(表示报单数据对象)

class Params(BaseParams):
    """参数映射模型 - 用于定义策略参数,这些参数将在无限易PythonGO窗口显示并可修改"""
    exchange: str = Field(default="", title="交易所代码")  # 交易所代码,字符串类型,默认空字符串,标题"交易所代码"(在窗口参数栏显示)
    instrument_id: str = Field(default="", title="合约代码")  # 合约代码,字符串类型,默认空字符串,标题"合约代码"
    order_price: int | float = Field(default=0, title="报单价格")  # 报单价格,整数或浮点数类型,默认0,标题"报单价格"
    order_volume: int = Field(default=1, title="报单手数")  # 报单手数,整数类型,默认1,标题"报单手数"
    order_direction: Literal["buy", "sell"] = Field(default="buy", title="报单方向")  # 报单方向,只能取"buy"或"sell",默认"buy",标题"报单方向"

class State(BaseState):
    """状态映射模型 - 用于定义策略状态,这些状态将在无限易PythonGO窗口状态栏实时显示"""
    order_id: int | None = Field(default=None, title="报单编号")  # 报单编号,整数或None类型,默认None,标题"报单编号"

class DemoTest(BaseStrategy):
    """我的第一个策略 - 策略类名必须与文件名一致,这是策略的描述信息"""
    def __init__(self) -> None:
        super().__init__()  # 调用父类BaseStrategy的初始化方法
        self.params_map = Params()  # 实例化参数映射模型,用于存储和获取策略参数
        self.state_map = State()  # 实例化状态映射模型,用于存储和获取策略状态

    def on_order(self, order: OrderData) -> None:
        super().on_order(order)  # 调用父类的on_order方法
        self.output("报单信息:", order)  # 输出报单信息到策略日志(注意:直接输出order会显示字典形式,因为已做处理)

    def on_start(self) -> None:
        super().on_start()  # 调用父类的on_start方法(策略启动时调用)
        # 使用send_order发送报单,参数从self.params_map中获取
        self.state_map.order_id = self.send_order(
            exchange=self.params_map.exchange,  # 从参数映射中获取交易所代码
            instrument_id=self.params_map.instrument_id,  # 从参数映射中获取合约代码
            volume=self.params_map.order_volume,  # 从参数映射中获取报单手数
            price=self.params_map.order_price,  # 从参数映射中获取报单价格
            order_direction=self.params_map.order_direction  # 从参数映射中获取报单方向
        )
        self.update_status_bar()  # 刷新状态栏,显示状态映射模型中的值(如报单编号)

    def on_stop(self) -> None:
        super().on_stop()  # 调用父类的on_stop方法(策略暂停时调用)
        self.output("我的第一个策略暂停了")  # 输出策略暂停信息到策略日志

Demo_DMA(官方案例,仅供软件学习使用,不作为投资建议)

# encoding: UTF-8  # 指定文件编码为 UTF-8

"""
双均线策略
last update: 2023年9月26日 16:27:36
"""

# 导入必要的模块
from typing import Dict, List  # 类型提示
from ctaTemplate import CtaTemplate, KLWidget  # CTA策略基类和K线可视化组件
from indicators import Indicators  # 技术指标计算模块
from utils import MinKLineGenerator  # K线生成器工具
from vtObject import KLineData, TickData  # 交易数据对象

class Demo_DMA(CtaTemplate):
    """仅供测试_可调节 K 线周期的双均线交易策略"""
    def __init__(self):
        super().__init__()  # 初始化基类
        
        # 参数映射表:将策略参数名映射为UI显示名称
        self.paramMap = {
            'exchange': '交易所',
            'vtSymbol': '合约',
            'fast_period': '快均线周期',
            'slow_period': '慢均线周期',
            'order_volume': '下单手数',
            'kline_style': 'K线类型',  # 如M5(5分钟K线)
            'pay_up': '超价'  # 买价加价/卖价减价的点数
        }

        # 变量映射表:将策略内部变量映射为UI显示名称
        self.varMap = {
            'trading': '交易中',
            'slow_ma': '慢均线',
            'fast_ma': '快均线',
            'pos': '持仓'  # 注意:实际代码中用的是position而非pos
        }
        
        # 创建K线可视化组件
        self.widgetClass = KLWidget
        self.widget: KLWidget = None  # K线界面对象
        
        # 初始化指标计算和K线生成工具
        self.indicators: Indicators = None  # 技术指标计算器
        self.kline_generator: MinKLineGenerator = None  # K线数据生成器

        # 策略核心参数(可配置)
        self.exchange = ""  # 交易所代码(如SHFE)
        self.vtSymbol = ""  # 合约代码(如rb2401)
        self.fast_period = 5  # 快均线周期(默认5根K线)
        self.slow_period = 20  # 慢均线周期(默认20根K线)
        self.order_volume = 1  # 每次下单手数
        self.kline_style = "M5"  # K线类型(M5=5分钟K线)
        self.pay_up = 0  # 价格超价(买价+pay_up,卖价-pay_up)

        # 均线指标缓存(当前K线和上一根K线)
        self.fast_ma = 0  # 当前K线的快均线值
        self.slow_ma = 0  # 当前K线的慢均线值
        self.pre_fast_ma = 0  # 上一根K线的快均线值
        self.pre_slow_ma = 0  # 上一根K线的慢均线值

        self.trading = False  # 交易状态(是否允许交易)

        # K线界面信号标记
        self.signal = 0  # 0=无信号,正数=买入信号,负数=卖出信号
        self.subSigs = []  # 子信号列表(用于多指标显示)

    @property
    def main_indicator(self) -> Dict[str, float]:
        """返回主图显示的指标(快慢均线)"""
        return {
            f"MA{self.fast_period}": self.fast_ma,  # 快均线名称
            f"MA{self.slow_period}": self.slow_ma  # 慢均线名称
        }

    @property
    def mainSigs(self) -> List[str]:
        """返回主图指标名称列表"""
        return list(self.main_indicator.keys())  # 返回['MA5', 'MA20']等

    def onTick(self, tick: TickData) -> None:
        """行情Tick回调函数(处理实时行情)"""
        super().onTick(tick)  # 调用基类处理
        
        # 仅当行情有有效价格时(避免无效数据)
        if tick.lastPrice or tick.askPrice1 or tick.bidPrice1:
            self.kline_generator.tick_to_kline(tick)  # 将Tick转为K线

    def calc_indicator(self) -> None:
        """计算技术指标(快慢均线)"""
        # 从指标计算器获取均线数组(最新值在末尾)
        slow_ma = self.indicators.sma(self.slow_period, array=True)  # 慢均线数组
        fast_ma = self.indicators.sma(self.fast_period, array=True)  # 快均线数组
        
        # 提取当前K线和上一根K线的均线值
        self.slow_ma, self.pre_slow_ma = slow_ma[-1], slow_ma[-2]
        self.fast_ma, self.pre_fast_ma = fast_ma[-1], fast_ma[-2]

    def calc_signal(self, kline: KLineData) -> None:
        """计算交易信号(基于均线交叉)"""
        hour = kline.datetime.hour  # 当前K线时间(小时)
        minute = kline.datetime.minute  # 当前K线时间(分钟)
        
        # 尾盘时段(14:40后)禁止交易并平仓
        self.end_of_day = hour == 14 and minute >= 40
        
        # 金叉:快线上穿慢线(买入信号)
        self.buy_signal = self.fast_ma > self.slow_ma and self.pre_fast_ma < self.pre_slow_ma
        
        # 死叉:快线下穿慢线(卖出信号)
        self.short_signal = self.fast_ma < self.slow_ma and self.pre_fast_ma > self.pre_slow_ma
        
        # 交易价格(使用K线收盘价)
        self.long_price = kline.close  # 买入开仓价
        self.short_price = kline.close  # 卖出开仓价

    def exec_signal(self) -> None:
        """执行交易信号(开仓/平仓)"""
        position = self.get_position(self.vtSymbol)  # 获取当前持仓
        
        # 如果有未成交订单,先撤单
        if self.orderID is not None:
            self.cancelOrder(self.orderID)
            self.orderID = None  # 清空订单ID

        self.signal = 0  # 重置信号标记

        # 情况1:无持仓且非尾盘时段
        if position.net_position == 0 and not self.end_of_day:
            # 卖出开仓信号(死叉)
            if self.short_signal:
                self.signal = abs(self.short_price)  # 标记卖出信号
                
                # 交易状态检查(防止策略未启动)
                if self.trading is False:
                    return
                
                # 下卖单(价格 = 卖价 - 超价)
                self.orderID = self.short(
                    price=(price := self.short_price - self.pay_up),
                    volume=self.order_volume
                )
                self.output(f'卖出开仓信号价格: {price}')
            
            # 买入开仓信号(金叉)
            elif self.buy_signal:
                self.signal = self.long_price  # 标记买入信号
                
                # 交易状态检查
                if self.trading is False:
                    return
                
                # 下买单(价格 = 买价 + 超价)
                self.orderID = self.buy(
                    price=(price := self.long_price + self.pay_up),
                    volume=self.order_volume
                )
                self.output(f'买入开仓信号价格: {price}')

        # 情况2:持有多头仓位且出现卖出信号
        elif position.net_position > 0 and self.short_signal:
            self.signal = -self.short_price  # 标记卖出平仓
            
            # 交易状态检查
            if self.trading is False:
                return
            
            # 平多仓(卖出平仓)
            self.orderID = self.auto_close_position(
                price=(price := self.short_price - self.pay_up),
                volume=position.net_position,  # 平全部仓位
                symbol=self.vtSymbol,
                exchange=self.exchange,
                order_direction="sell"
            )
            self.output(f'卖出平仓信号价格: {price}')

        # 情况3:持有空头仓位且出现买入信号
        elif position.net_position < 0 and self.buy_signal:
            self.signal = self.long_price  # 标记买入平仓
            
            # 交易状态检查
            if self.trading is False:
                return
            
            # 平空仓(买入平仓)
            self.orderID = self.auto_close_position(
                price=(price := self.long_price + self.pay_up),
                volume=abs(position.net_position),  # 平全部空仓
                symbol=self.vtSymbol,
                exchange=self.exchange,
                order_direction="buy"
            )
            self.output(f'买入平仓信号价格: {price}')

    def onInit(self):
        """策略初始化(启动前调用)"""
        super().onInit()  # 调用基类初始化
        self.getGui()  # 获取GUI组件(用于显示K线图)

    def onTrade(self, trade, log=True):
        """成交回调(成交后处理)"""
        super().onTrade(trade, log)  # 调用基类处理

    def callback(self, kline: KLineData) -> None:
        """K线数据回调函数(策略核心逻辑)"""
        self.calc_indicator()  # 计算均线指标
        
        self.calc_signal(kline)  # 生成交易信号
        
        self.exec_signal()  # 执行交易信号
        
        self.putEvent()  # 更新策略状态事件
        
        # 发送K线数据到可视化组件
        self.widget.recv_kline({
            'bar': kline,  # 当前K线数据
            'sig': self.signal,  # 交易信号
            **self.main_indicator  # 主图指标(快慢均线)
        })

    def real_time_callback(self, kline: KLineData) -> None:
        """实时K线回调(用于K线图实时刷新)"""
        self.calc_indicator()  # 仅计算指标
        
        # 仅刷新K线图(不触发交易信号)
        self.widget.recv_kline({
            'bar': kline,
            'sig': 0,  # 无交易信号
            **self.main_indicator
        })

    def onStart(self):
        """策略启动(初始化K线生成器)"""
        # 创建K线生成器(接收实时Tick数据)
        self.kline_generator = MinKLineGenerator(
            real_time_callback=self.real_time_callback,  # 实时K线回调
            callback=self.callback,  # 策略核心逻辑回调
            exchange=self.exchange,  # 交易所
            instrument=self.vtSymbol,  # 合约
            style=self.kline_style  # K线周期(如M5)
        )
        
        super().onStart()  # 调用基类启动

    def onStop(self):
        """策略停止(清理资源)"""
        if self.kline_generator:
            self.kline_generator.stop_push_scheduler()  # 停止K线推送
        
        super().onStop()  # 调用基类停止

==================== End