用大模型工具代码文件写的一个策略编写教程

vnpy CTA 策略脚本完整编写指南

本指南基于你本地两份真实源码编写，不包含任何臆造内容：

D:\Desktop\vnpy（vnpy 主框架源码，版本 4.4.0）

D:\Desktop\vnpy_ctastrategy（CTA 策略模块源码，版本 1.4.1）

经核对，这两份源码与安装路径 C:\veighna_studio\Lib\site-packages 下的 vnpy（4.4.0）与 vnpy_ctastrategy（1.4.1）内容完全一致，因此本文档描述的就是你当前环境里正在运行的实现。

文中每一个函数签名、参数默认值、枚举取值，均直接摘自上述源码文件，并注明来源文件，方便你对照查阅。

核心概念：什么是 CTA 策略
源码文件结构与职责
一个最小可运行策略的骨架
CtaTemplate 基类完全解析
数据结构详解（BarData / TickData / OrderData / TradeData / StopOrder / ContractData 等）
常量枚举详解（Direction / Offset / Interval / Exchange / Status / OrderType 等）
BarGenerator：K 线合成器完全解析
ArrayManager：技术指标计算器完全解析
CtaSignal 与 TargetPosTemplate
官方示例策略逐行解读
回测引擎 BacktestingEngine
实盘引擎 CtaEngine 与策略生命周期
编写策略的完整步骤清单
常见陷阱与最佳实践

1. 核心概念：什么是 CTA 策略

在 vnpy 中，一个 CTA（Commodity Trading Advisor）策略本质上是一个继承自 CtaTemplate 的 Python 类。框架通过「事件回调」的方式驱动策略：

行情到达 → 引擎调用策略的 on_tick() / on_bar()
委托状态变化 → 引擎调用策略的 on_order()
成交回报到达 → 引擎调用策略的 on_trade()
停止单状态变化 → 引擎调用策略的 on_stop_order()

你在这些回调里编写交易逻辑，并通过 buy() / sell() / short() / cover() 等方法下单。

策略的三大类属性（写在类体里，不是 __init__ 里）：

属性	类型	作用
`author`	`str`	作者名，展示在界面上
`parameters`	`list[str]`	策略参数名列表，可在界面上修改、会参与回测参数优化
`variables`	`list[str]`	策略状态变量名列表，会实时展示在界面、会持久化到磁盘

这三者定义在 template.py 中：

class CtaTemplate(ABC):
    author: str = ""
    parameters: list = []
    variables: list = []

关键区别：parameters 里的变量属于「外部可调参数」，variables 里的变量属于「内部状态」。前者改动后重新初始化生效；后者在实盘中每次停止策略时会写入 cta_strategy_data.json，下次初始化会回读恢复（见第 12 节 _init_strategy 与 sync_strategy_data）。

2. 源码文件结构与职责

2.1 `D:\Desktop\vnpy\trader`（主框架）

文件	与你写策略相关的职责
`object.py`	定义 `BarData`、`TickData`、`OrderData`、`TradeData`、`ContractData`、`PositionData`、`AccountData`、`OrderRequest` 等所有数据结构
`constant.py`	定义 `Direction`、`Offset`、`Interval`、`Exchange`、`Status`、`OrderType`、`Product` 等枚举
`utility.py`	定义 `BarGenerator`、`ArrayManager`、`round_to`、`extract_vt_symbol` 等工具

2.2 `D:\Desktop\vnpy_ctastrategy`（CTA 策略模块）

文件	职责
`template.py`	定义 `CtaTemplate`、`CtaSignal`、`TargetPosTemplate` 三个基类（写策略的核心）
`base.py`	定义 `StopOrder` 数据类、`StopOrderStatus`、`EngineType`、`BacktestingMode`、`INTERVAL_DELTA_MAP` 等
`engine.py`	定义 `CtaEngine`（实盘引擎）
`backtesting.py`	定义 `BacktestingEngine`（回测引擎）
`strategies/`	官方示例策略目录
`__init__.py`	对外导出的公共接口

2.3 策略文件应该放在哪

engine.py 的 load_strategy_class() 会从两个目录扫描 .py/.pyd/.so 文件并加载其中的策略类：

def load_strategy_class(self) -> None:
    path1: Path = Path(__file__).parent.joinpath("strategies")   # 安装包自带目录
    self.load_strategy_class_from_folder(path1, "vnpy_ctastrategy.strategies")

    path2: Path = Path.cwd().joinpath("strategies")              # 程序运行目录下的 strategies
    self.load_strategy_class_from_folder(path2, "strategies")

也就是说：把你自己的策略 .py 文件放到 vnpy 启动目录下的 strategies/ 文件夹（推荐做法），或直接放进 vnpy_ctastrategy/strategies/ 里，程序启动时就能自动识别加载。加载逻辑会遍历模块中所有继承自 CtaTemplate（且不等于 CtaTemplate、TargetPosTemplate 本身）的类。

3. 一个最小可运行策略的骨架

下面是一个能直接放进 strategies/ 运行的最小策略（仅演示结构，不代表盈利逻辑）：

from vnpy_ctastrategy import (
    CtaTemplate,
    StopOrder,
    TickData,
    BarData,
    TradeData,
    OrderData,
    BarGenerator,
    ArrayManager,
)


class MyFirstStrategy(CtaTemplate):
    """"""
    author = "你的名字"

    # ---- 参数：界面上可修改 ----
    fast_window: int = 10
    slow_window: int = 20

    # ---- 变量：内部状态 ----
    fast_ma: float = 0.0
    slow_ma: float = 0.0

    parameters = ["fast_window", "slow_window"]
    variables = ["fast_ma", "slow_ma"]

    def on_init(self) -> None:
        """初始化时调用一次"""
        self.write_log("策略初始化")
        self.bg: BarGenerator = BarGenerator(self.on_bar)
        self.am: ArrayManager = ArrayManager()
        self.load_bar(10)          # 加载 10 天历史数据用于预热

    def on_start(self) -> None:
        """启动时调用一次"""
        self.write_log("策略启动")

    def on_stop(self) -> None:
        """停止时调用一次"""
        self.write_log("策略停止")

    def on_tick(self, tick: TickData) -> None:
        """每个 tick 到达时调用"""
        self.bg.update_tick(tick)   # tick 合成 1 分钟 K 线

    def on_bar(self, bar: BarData) -> None:
        """每根 1 分钟 K 线完成时调用"""
        self.am.update_bar(bar)
        if not self.am.inited:
            return

        self.fast_ma = self.am.sma(self.fast_window)
        self.slow_ma = self.am.sma(self.slow_window)

        cross_over = self.fast_ma > self.slow_ma
        cross_below = self.fast_ma < self.slow_ma

        if self.pos == 0:
            if cross_over:
                self.buy(bar.close_price, 1)
            elif cross_below:
                self.short(bar.close_price, 1)
        elif self.pos > 0 and cross_below:
            self.sell(bar.close_price, 1)
        elif self.pos < 0 and cross_over:
            self.cover(bar.close_price, 1)

        self.put_event()

    def on_order(self, order: OrderData) -> None:
        pass

    def on_trade(self, trade: TradeData) -> None:
        self.put_event()

    def on_stop_order(self, stop_order: StopOrder) -> None:
        pass

这个骨架已经展示了全部核心要素，下面逐项展开说明。

4. CtaTemplate 基类完全解析

来源：vnpy_ctastrategy/template.py。下面按「自动生成的实例属性 → 生命周期回调 → 交易函数 → 辅助函数」四部分逐个说明。

4.1 构造函数 `init`

def __init__(self, cta_engine, strategy_name, vt_symbol, setting) -> None:
    self.cta_engine = cta_engine
    self.strategy_name = strategy_name
    self.vt_symbol = vt_symbol

    self.inited: bool = False
    self.trading: bool = False
    self.pos: float = 0

    self.variables = copy(self.variables)
    self.variables.insert(0, "inited")
    self.variables.insert(1, "trading")
    self.variables.insert(2, "pos")

    self.update_setting(setting)

你通常不需要重写 __init__（基类已处理引擎绑定与参数注入）。构造函数帮你建立以下实例属性：

实例属性	类型	含义
`self.cta_engine`	引擎对象	实盘是 `CtaEngine`，回测是 `BacktestingEngine`
`self.strategy_name`	`str`	策略实例名（界面上创建实例时填的名字）
`self.vt_symbol`	`str`	合约代码，格式 `"rb2410.SHFE"`（品种.交易所）
`self.inited`	`bool`	是否已初始化
`self.trading`	`bool`	是否正在交易（启动后为 True）
`self.pos`	`float`	当前持仓（正=多头，负=空头，由引擎自动维护）

注意两点：

self.variables = copy(self.variables) 是为了避免同一个策略类创建多个实例时，向类级 variables 列表重复插入导致的污染（注释原文如此）。
self.pos 由引擎在收到成交回报时自动更新（见 engine.py 的 process_trade_event：Direction.LONG 时 pos += volume，否则 pos -= volume），你不应该手动改它。

4.2 参数与状态相关方法

`update_setting(setting: dict) -> None`

def update_setting(self, setting: dict) -> None:
    for name in self.parameters:
        if name in setting:
            setattr(self, name, setting[name])

用途：把 setting 字典中与 parameters 同名的键值写入实例属性。界面上「修改参数」、回测时传入参数，最终都通过它生效。

`get_class_parameters() -> dict`（classmethod）

@classmethod
def get_class_parameters(cls) -> dict:
    class_parameters: dict = {}
    for name in cls.parameters:
        class_parameters[name] = getattr(cls, name)
    return class_parameters

用途：返回策略类的默认参数字典（即类体里写的默认值），用于界面展示默认参数。

`get_parameters() -> dict`

def get_parameters(self) -> dict:
    strategy_parameters: dict = {}
    for name in self.parameters:
        strategy_parameters[name] = getattr(self, name)
    return strategy_parameters

用途：返回当前实例的参数字典（可能已被 setting 覆盖）。

`get_variables() -> dict`

def get_variables(self) -> dict:
    strategy_variables: dict = {}
    for name in self.variables:
        strategy_variables[name] = getattr(self, name)
    return strategy_variables

用途：返回当前实例所有 variables（含自动追加的 inited/trading/pos）的状态字典。实盘停止时就是用它做持久化。

`get_data() -> dict`

def get_data(self) -> dict:
    return {
        "strategy_name": self.strategy_name,
        "vt_symbol": self.vt_symbol,
        "class_name": self.__class__.__name__,
        "author": self.author,
        "parameters": self.get_parameters(),
        "variables": self.get_variables(),
    }

用途：返回策略的完整快照，供界面刷新显示用。

4.3 生命周期回调函数

这些函数在策略的不同阶段被引擎调用，是写交易逻辑的入口。

`on_init(self) -> None`（抽象方法，必须实现）

@abstractmethod
def on_init(self) -> None:
    return

调用时机：用户点击「初始化」或回测开始时调用一次。

典型用法：

创建 BarGenerator、ArrayManager 等指标对象
计算派生参数（如 RSI 上下阈值）
调用 self.load_bar(days) 加载历史数据做指标预热
调用 self.write_log() 输出日志

注意：on_init 是 @abstractmethod，继承 CtaTemplate 的子类必须实现它，否则无法实例化。

`on_start(self) -> None`

def on_start(self) -> None:
    return

调用时机：用户点击「启动」时调用一次，且必须先完成初始化才能启动（引擎会检查 strategy.inited）。

典型用法：输出日志、self.put_event() 刷新界面。

`on_stop(self) -> None`

def on_stop(self) -> None:
    return

调用时机：用户点击「停止」时调用一次。

注意：引擎在调用 on_stop 之后会把 trading 置为 False 并自动撤销该策略的所有未成交委托（见 stop_strategy 中的 self.cancel_all(strategy)），无需你在 on_stop 里手动撤单。

`on_tick(self, tick: TickData) -> None`

def on_tick(self, tick: TickData) -> None:
    return

调用时机：订阅的合约每来一个 tick 调用一次（实盘中）；tick 回测模式下每回放一个 tick 调用一次。

典型用法：self.bg.update_tick(tick) 把 tick 喂给 BarGenerator 合成 K 线；或直接使用 tick 数据做高频逻辑。

`on_bar(self, bar: BarData) -> None`

def on_bar(self, bar: BarData) -> None:
    return

调用时机：每完成一根 K 线调用一次。既可能是 BarGenerator 合成的 K 线，也可能是实盘直接订阅的 K 线，或回测中回放的 K 线。

典型用法：核心交易逻辑所在。常见模板：

def on_bar(self, bar):
    self.cancel_all()            # 可选：先撤掉上一根 K 线挂的未成交单
    self.am.update_bar(bar)      # 更新指标容器
    if not self.am.inited:       # 指标尚未就绪则返回
        return
    # ... 计算信号、下单 ...
    self.put_event()             # 刷新界面

`on_trade(self, trade: TradeData) -> None`

def on_trade(self, trade: TradeData) -> None:
    return

调用时机：每次成交回报到达时调用。

注意：引擎在调用 on_trade 之前已经更新了 self.pos（见 process_trade_event），所以你在 on_trade 里读到的 self.pos 已是最新持仓。

`on_order(self, order: OrderData) -> None`

def on_order(self, order: OrderData) -> None:
    return

调用时机：委托状态变化（提交、未成交、部分成交、全部成交、撤单、拒单）时调用。配合 order.status、order.is_active() 可以跟踪委托状态。

`on_stop_order(self, stop_order: StopOrder) -> None`

def on_stop_order(self, stop_order: StopOrder) -> None:
    return

调用时机：本地/服务器停止单状态变化（等待中 → 已触发 / 已撤销）时调用。参数是 StopOrder 对象（见 5.6 节）。

4.4 交易函数

这四个函数是下单的唯一入口，它们内部都调用 send_order()，区别仅在于封装的 direction 和 offset 组合：

函数	方向	开平	语义
`buy(price, volume, ...)`	LONG	OPEN	买入开多
`sell(price, volume, ...)`	SHORT	CLOSE	卖出平多
`short(price, volume, ...)`	SHORT	OPEN	卖出开空
`cover(price, volume, ...)`	LONG	CLOSE	买入平空

源码对应关系（节选）：

def buy(self, price, volume, stop=False, lock=False, net=False):
    return self.send_order(Direction.LONG, Offset.OPEN, price, volume, stop, lock, net)

def sell(self, price, volume, stop=False, lock=False, net=False):
    return self.send_order(Direction.SHORT, Offset.CLOSE, price, volume, stop, lock, net)

def short(self, price, volume, stop=False, lock=False, net=False):
    return self.send_order(Direction.SHORT, Offset.OPEN, price, volume, stop, lock, net)

def cover(self, price, volume, stop=False, lock=False, net=False):
    return self.send_order(Direction.LONG, Offset.CLOSE, price, volume, stop, lock, net)

参数说明（四个函数一致）：

参数	类型	说明
`price`	`float`	委托价格
`volume`	`float`	委托数量（手）
`stop`	`bool`	是否为止损单（True 时发停止单，见下文）
`lock`	`bool`	是否锁仓模式（见 14 节）
`net`	`bool`	是否净持仓模式（见 14 节）

返回值：list[str]，即本次委托生成的 vt_orderid 列表（可能为空）。实盘里若 self.trading == False 则返回空列表；回测里每个订单生成一个 vt_orderid。

stop=True 的含义：当 stop=True 时，引擎会下「停止单」（条件单）而不是限价单。触发规则由引擎实现：

实盘（CtaEngine.check_stop_order）：当最新价 last_price >= 触发价（多头）或 <= 触发价（空头）时触发，随后以对手五档价（ask_price_5/bid_price_5，封涨跌停则用涨跌停价）发限价单。
回测（BacktestingEngine.cross_stop_order）：K 线模式下多头用 bar.high_price 判断是否触达、空头用 bar.low_price，成交价取 max(触发价, open_price) / min(触发价, open_price)。

停止单最典型的用途就是「止损/移动止损」，例如官方 AtrRsiStrategy：

elif self.pos > 0:
    long_stop = self.intra_trade_high * (1 - self.trailing_percent / 100)
    self.sell(long_stop, abs(self.pos), stop=True)

`send_order(direction, offset, price, volume, stop=False, lock=False, net=False) -> list`

def send_order(self, direction, offset, price, volume, stop=False, lock=False, net=False):
    if self.trading:
        vt_orderids = self.cta_engine.send_order(self, direction, offset, price, volume, stop, lock, net)
        return vt_orderids
    else:
        return []

用途：通用下单函数，比 buy/sell/short/cover 更底层，允许你直接指定 direction 和 offset 的任意组合。

参数	类型	说明
`direction`	`Direction`	方向枚举
`offset`	`Offset`	开平枚举
`price`	`float`	委托价格
`volume`	`float`	委托数量
`stop` / `lock` / `net`	`bool`	同 `buy`

重要：只有当 self.trading == True 时才真正发单，否则静默返回 []。这是防止策略未启动时误发单的保护。

`cancel_order(vt_orderid: str) -> None`

def cancel_order(self, vt_orderid: str) -> None:
    if self.trading:
        self.cta_engine.cancel_order(self, vt_orderid)

用途：撤销指定 vt_orderid 的委托（既支持限价单也支持本地停止单——引擎内部通过 vt_orderid 是否以 "STOP" 前缀开头来区分，见 STOPORDER_PREFIX = "STOP"）。

`cancel_all() -> None`

def cancel_all(self) -> None:
    if self.trading:
        self.cta_engine.cancel_all(self)

用途：撤销该策略发出的所有未成交委托（限价单 + 停止单）。很多策略在每根新 K 线开头先 cancel_all() 再重新挂单。

4.5 辅助函数

`write_log(msg: str) -> None`

def write_log(self, msg: str) -> None:
    self.cta_engine.write_log(msg, self)

用途：写一条日志。实盘中显示在「CTA策略」的日志监控里，回测中记录到回测日志。引擎会自动在消息前加上 [策略名] 前缀（见 CtaEngine.write_log）。

`get_engine_type() -> EngineType`

def get_engine_type(self) -> EngineType:
    return cast(EngineType, self.cta_engine.get_engine_type())

用途：返回当前引擎类型 EngineType.LIVE（实盘）或 EngineType.BACKTESTING（回测）。用于写「回测与实盘行为不同」的分支逻辑。

`get_pricetick() -> float`

def get_pricetick(self) -> float:
    return cast(float, self.cta_engine.get_pricetick(self))

用途：返回当前合约的最小价格变动单位（pricetick）。实盘中来自合约信息，回测中来自 set_parameters 传入的值。

`get_size() -> int`

def get_size(self) -> int:
    return cast(int, self.cta_engine.get_size(self))

用途：返回当前合约的合约乘数（size）。实盘中来自合约信息，回测中来自 set_parameters 传入的值。

`load_bar(days, interval=Interval.MINUTE, callback=None, use_database=False) -> None`

def load_bar(self, days, interval=Interval.MINUTE, callback=None, use_database=False) -> None:
    if not callback:
        callback = self.on_bar

    bars = self.cta_engine.load_bar(self.vt_symbol, days, interval, callback, use_database)

    for bar in bars:
        callback(bar)

用途：加载历史 K 线数据用于「预热」指标（让 ArrayManager 攒够数据达到 inited 状态）。

参数	说明
`days`	往回加载多少天
`interval`	K 线周期，默认 `Interval.MINUTE`
`callback`	每条数据回调用哪个函数，默认 `self.on_bar`
`use_database`	`False`（默认）优先走网关/datafeed，`True` 强制走数据库（见 `CtaEngine.load_bar`）

调用方式：在 on_init 里调用 self.load_bar(10)，它会同步地把这 10 天的历史 bar 一条条喂给 self.on_bar，从而完成 BarGenerator/ArrayManager 的预热。注意实盘与回测取数区间不同：实盘取「现在往前 N 天」，回测取「回测开始时间往前 N 天」（见 BacktestingEngine.load_bar 里 init_start = self.start - timedelta(days=days)）。

`load_tick(days: int) -> None`

def load_tick(self, days: int) -> None:
    ticks = self.cta_engine.load_tick(self.vt_symbol, days, self.on_tick)
    for tick in ticks:
        self.on_tick(tick)

用途：加载历史 tick 数据用于预热（类似 load_bar，但喂给 on_tick）。

`put_event() -> None`

def put_event(self) -> None:
    if self.inited:
        self.cta_engine.put_strategy_event(self)

用途：把策略最新状态推送到界面刷新（让 variables 里的数值实时更新到 GUI）。在 on_bar/on_trade 等回调末尾调用一次即可，让界面看到 pos、指标值等变化。

`send_notification(msg: str) -> None` / `send_email`

def send_notification(self, msg: str) -> None:
    if self.inited:
        self.cta_engine.send_notification(msg, self)

send_email = send_notification

用途：把消息推送到所有已配置的通知渠道（邮件、微信等，取决于主引擎的通知模块）。注意：send_email 只是 send_notification 的别名（源码里 send_email = send_notification），并不是独立的邮件函数——是否通过邮件发出，取决于通知引擎的渠道配置。

`sync_data() -> None`

def sync_data(self) -> None:
    if self.trading:
        self.cta_engine.sync_strategy_data(self)

用途：把 variables 状态立即写入磁盘（cta_strategy_data.json）。正常情况下引擎会在成交后自动同步，你也可以在关键节点手动调用。

5. 数据结构详解

来源：vnpy/trader/object.py、vnpy_ctastrategy/base.py。所有数据类都用 @dataclass 定义，且继承自 BaseData（含 gateway_name 字段）。

5.1 BarData（K 线数据）

@dataclass
class BarData(BaseData):
    symbol: str
    exchange: Exchange
    datetime: Datetime

    interval: Interval | None = None
    volume: float = 0
    turnover: float = 0
    open_interest: float = 0
    open_price: float = 0
    high_price: float = 0
    low_price: float = 0
    close_price: float = 0

字段	类型	含义
`symbol`	`str`	合约代码（如 `rb2410`）
`exchange`	`Exchange`	交易所枚举
`datetime`	`datetime`	K 线时间
`interval`	`Interval`	K 线周期
`volume`	`float`	成交量
`turnover`	`float`	成交额
`open_interest`	`float`	持仓量
`open_price`	`float`	开盘价
`high_price`	`float`	最高价
`low_price`	`float`	最低价
`close_price`	`float`	收盘价
`vt_symbol`	`str`	自动生成：`f"{symbol}.{exchange.value}"`（如 `rb2410.SHFE`）

策略中最常使用的是 bar.close_price、bar.high_price、bar.low_price、bar.open_price、bar.datetime。

5.2 TickData（Tick 数据）

@dataclass
class TickData(BaseData):
    symbol: str
    exchange: Exchange
    datetime: Datetime

    name: str = ""
    volume: float = 0
    turnover: float = 0
    open_interest: float = 0
    last_price: float = 0
    last_volume: float = 0
    limit_up: float = 0
    limit_down: float = 0

    open_price: float = 0
    high_price: float = 0
    low_price: float = 0
    pre_close: float = 0

    bid_price_1: float = 0
    # ... bid_price_2 ~ bid_price_5 同理
    ask_price_1: float = 0
    # ... ask_price_2 ~ ask_price_5 同理
    bid_volume_1: float = 0
    # ... bid_volume_2 ~ bid_volume_5 同理
    ask_volume_1: float = 0
    # ... ask_volume_2 ~ ask_volume_5 同理

    localtime: Datetime | None = None

常用字段	含义
`last_price`	最新成交价
`last_volume`	最新成交量（单笔）
`volume`	当日累计成交量
`turnover`	当日累计成交额
`open_interest`	持仓量
`open_price` / `high_price` / `low_price`	当日开盘/最高/最低
`pre_close`	昨收
`limit_up` / `limit_down`	涨停价 / 跌停价
`bid_price_1`~`bid_price_5`	买一 ~ 买五价
`ask_price_1`~`ask_price_5`	卖一 ~ 卖五价
`bid_volume_1`~`bid_volume_5`	买一 ~ 买五量
`ask_volume_1`~`ask_volume_5`	卖一 ~ 卖五量

这些字段是官方 TargetPosTemplate.send_new_order 中用来「追价下单」的依据（用 ask_price_1 + tick_add 作为买入价、bid_price_1 - tick_add 作为卖出价）。

5.3 OrderData（委托数据）

@dataclass
class OrderData(BaseData):
    symbol: str
    exchange: Exchange
    orderid: str

    type: OrderType = OrderType.LIMIT
    direction: Direction | None = None
    offset: Offset = Offset.NONE
    price: float = 0
    volume: float = 0
    traded: float = 0
    status: Status = Status.SUBMITTING
    datetime: Datetime | None = None
    reference: str = ""

字段	含义
`orderid`	本地委托号
`vt_orderid`	全局委托号：`f"{gateway_name}.{orderid}"`
`type`	委托类型（`OrderType`）
`direction`	方向
`offset`	开平
`price`	委托价
`volume`	委托量
`traded`	已成交量
`status`	委托状态（`Status`）
`reference`	策略下单时写入的参考（CTA 引擎写 `"CtaStrategy_策略名"`）

OrderData 的两个方法：

def is_active(self) -> bool:
    return self.status in ACTIVE_STATUSES
# ACTIVE_STATUSES = {Status.SUBMITTING, Status.NOTTRADED, Status.PARTTRADED}

is_active() 返回该委托是否仍「活跃」（提交中/未成交/部分成交）。配合 on_order 可用于跟踪活动委托数。

def create_cancel_request(self) -> "CancelRequest":
    req = CancelRequest(orderid=self.orderid, symbol=self.symbol, exchange=self.exchange)
    return req

create_cancel_request() 从委托生成一个撤单请求（框架内部撤单时使用）。

5.4 TradeData（成交数据）

@dataclass
class TradeData(BaseData):
    symbol: str
    exchange: Exchange
    orderid: str
    tradeid: str
    direction: Direction | None = None

    offset: Offset = Offset.NONE
    price: float = 0
    volume: float = 0
    datetime: Datetime | None = None

字段	含义
`tradeid`	成交编号
`vt_tradeid`	全局成交号：`f"{gateway_name}.{tradeid}"`
`price`	成交价
`volume`	成交量
`direction`	方向
`offset`	开平

一个委托可能分多笔成交，因此一个 OrderData 可对应多个 TradeData。

5.5 ContractData（合约数据）

@dataclass
class ContractData(BaseData):
    symbol: str
    exchange: Exchange
    name: str
    product: Product
    size: float
    pricetick: float

    min_volume: float = 1
    max_volume: float | None = None
    stop_supported: bool = False
    net_position: bool = False
    history_data: bool = False
    # ... 期权相关字段略 ...

与策略相关的重要字段：

字段	含义
`size`	合约乘数（如螺纹钢为 10）
`pricetick`	最小变动价位（如螺纹钢为 1）
`min_volume`	最小下单量
`stop_supported`	交易所/网关是否支持服务器停止单
`net_position`	是否为净持仓模式
`history_data`	网关是否提供历史数据

策略中通过 self.get_pricetick() 和 self.get_size() 间接获取 pricetick 与 size。

5.6 StopOrder（停止单）

来源：vnpy_ctastrategy/base.py。

@dataclass
class StopOrder:
    vt_symbol: str
    direction: Direction
    offset: Offset
    price: float
    volume: float
    stop_orderid: str
    strategy_name: str
    datetime: datetime
    lock: bool = False
    net: bool = False
    vt_orderids: list = field(default_factory=list)
    status: StopOrderStatus = StopOrderStatus.WAITING

字段	含义
`stop_orderid`	停止单编号（本地停止单以 `STOP.` 开头）
`price`	触发价
`status`	`StopOrderStatus.WAITING` / `TRIGGERED` / `CANCELLED`
`vt_orderids`	触发后实际生成的限价单 id 列表

StopOrderStatus 枚举：

class StopOrderStatus(Enum):
    WAITING = "等待中"
    CANCELLED = "已撤销"
    TRIGGERED = "已触发"

5.7 其他数据类（了解即可）

PositionData：持仓数据（volume 持仓量、frozen 冻结量、price 均价、pnl 盈亏、yd_volume 昨仓）。
AccountData：账户数据（balance 余额、frozen 冻结、available 可用）。
OrderRequest / CancelRequest / SubscribeRequest / HistoryRequest：内部请求对象，策略中一般不直接使用。

6. 常量枚举详解

来源：vnpy/trader/constant.py。

6.1 Direction（方向）

class Direction(Enum):
    LONG = "多"
    SHORT = "空"
    NET = "净"

用法：from vnpy.trader.constant import Direction，然后 Direction.LONG、Direction.SHORT。NET 用于净持仓概念的接口。

6.2 Offset（开平）

class Offset(Enum):
    NONE = ""
    OPEN = "开"
    CLOSE = "平"
    CLOSETODAY = "平今"
    CLOSEYESTERDAY = "平昨"

用法：Offset.OPEN（开仓）、Offset.CLOSE（平仓）。CLOSETODAY/CLOSEYESTERDAY 用于区分平今/平昨（上期所、能源中心等有平今指令的交易所）。策略层通常只用 buy/sell/short/cover 间接对应 OPEN/CLOSE，无需手动指定。

6.3 Interval（K 线周期）

class Interval(Enum):
    MINUTE = "1m"
    HOUR = "1h"
    DAILY = "d"
    WEEKLY = "w"
    TICK = "tick"

用法：self.load_bar(10, Interval.MINUTE)。注意 Interval 的值是字符串（"1m" 等），对应数据库/数据服务中的周期标识。

6.4 Exchange（交易所）

class Exchange(Enum):
    # 中国
    CFFEX = "CFFEX"   # 中金所
    SHFE = "SHFE"     # 上期所
    CZCE = "CZCE"     # 郑商所
    DCE = "DCE"       # 大商所
    INE = "INE"       # 上海国际能源中心
    GFEX = "GFEX"     # 广期所
    SSE = "SSE"       # 上交所
    SZSE = "SZSE"     # 深交所
    BSE = "BSE"       # 北交所
    SHHK = "SHHK"     # 沪港通
    SZHK = "SZHK"     # 深港通
    SGE = "SGE"       # 上金所
    WXE = "WXE"
    CFETS = "CFETS"
    XBOND = "XBOND"
    # 全球
    SMART = "SMART"; NYSE = "NYSE"; NASDAQ = "NASDAQ"; ...
    # 特殊
    LOCAL = "LOCAL"
    GLOBAL = "GLOBAL"

（完整枚举包含大量海外交易所，此处仅列与中国市场相关的主要项，全部定义见 constant.py。）

vt_symbol 的构造就是 f"{symbol}.{exchange.value}"，例如螺纹钢主力 "rb2410.SHFE"。创建策略实例时，引擎会校验 vt_symbol 是否含 . 且交易所后缀是否合法。

6.5 Status（委托状态）

class Status(Enum):
    SUBMITTING = "提交中"
    NOTTRADED = "未成交"
    PARTTRADED = "部分成交"
    ALLTRADED = "全部成交"
    CANCELLED = "已撤销"
    REJECTED = "拒单"

配合 order.is_active()：SUBMITTING/NOTTRADED/PARTTRADED 属于活跃状态。

6.6 OrderType（委托类型）

class OrderType(Enum):
    LIMIT = "限价"
    MARKET = "市价"
    STOP = "STOP"
    FAK = "FAK"
    FOK = "FOK"
    RFQ = "询价"
    ETF = "ETF"

策略层通过 buy(..., stop=True) 触发 STOP 类型，其余类型一般由底层网关处理。

6.7 Product（产品类型）

class Product(Enum):
    EQUITY = "股票"
    FUTURES = "期货"
    OPTION = "期权"
    INDEX = "指数"
    FOREX = "外汇"
    SPOT = "现货"
    ETF = "ETF"
    BOND = "债券"
    WARRANT = "权证"
    SPREAD = "价差"
    FUND = "基金"
    CFD = "CFD"
    SWAP = "互换"

6.8 Currency（货币）

class Currency(Enum):
    USD = "USD"
    HKD = "HKD"
    CNY = "CNY"
    CAD = "CAD"

7. BarGenerator：K 线合成器完全解析

来源：vnpy/trader/utility.py。BarGenerator 的作用是把低周期数据合成为高周期 K 线：

用 tick 合成 1 分钟 K 线
用 1 分钟 K 线合成 N 分钟 / N 小时 K 线
合成日 K 线

7.1 构造函数

def __init__(self, on_bar, window=0, on_window_bar=None, interval=Interval.MINUTE, daily_end=None):
    self.bar = None
    self.on_bar = on_bar

    self.interval = interval
    self.interval_count = 0

    self.hour_bar = None
    self.daily_bar = None

    self.window = window
    self.window_bar = None
    self.on_window_bar = on_window_bar

    self.last_tick = None

    self.daily_end = daily_end
    if self.interval == Interval.DAILY and not self.daily_end:
        raise RuntimeError("合成日K线必须传入每日收盘时间")

参数	类型	说明
`on_bar`	`Callable`	合成 1 分钟 K 线完成后的回调（即策略的 `on_bar`）
`window`	`int`	要合成的窗口大小（如 5 表示合成 5 分钟 K 线）；0 表示不合成窗口
`on_window_bar`	`Callable`	合成 `window` 周期 K 线完成后的回调（如 `on_5min_bar`）
`interval`	`Interval`	窗口合成类型，默认 `Interval.MINUTE`（合成 N 分钟）；也可用 `HOUR`/`DAILY`
`daily_end`	`time`	仅当 `interval == Interval.DAILY` 时必填，日线收盘判定时间

三种常见用法：

# 1. 仅用 tick 合成 1 分钟 K 线（最常用）
self.bg = BarGenerator(self.on_bar)

# 2. 用 1 分钟 K 线合成 5 分钟 K 线
self.bg = BarGenerator(self.on_bar, 5, self.on_5min_bar)

# 3. 用 1 分钟 K 线合成 1 小时 K 线
self.bg = BarGenerator(self.on_bar, 1, self.on_hour_bar, Interval.HOUR)

7.2 update_tick(tick)

def update_tick(self, tick: TickData) -> None:
    # 过滤 last_price 为 0 的 tick
    if not tick.last_price:
        return
    # 若分钟变化，则先把上一根 1 分钟 K 线推给 on_bar，再新建当前 K 线
    ...

用途：把 tick 喂入，内部按分钟切分合成 1 分钟 K 线。当一根 1 分钟 K 线完成时调用 self.on_bar(bar)。这是 tick 驱动策略的标准入口：on_tick 里 self.bg.update_tick(tick)。

关键细节（源码原文注释）：

合成 N 分钟 K 线时，window（N）必须是能整除 60 的数：2, 3, 5, 6, 10, 15, 20, 30。
合成 N 小时 K 线时，window（N）可以是任意整数。

7.3 update_bar(bar)

def update_bar(self, bar: BarData) -> None:
    if self.interval == Interval.MINUTE:
        self.update_bar_minute_window(bar)
    elif self.interval == Interval.HOUR:
        self.update_bar_hour_window(bar)
    else:
        self.update_bar_daily_window(bar)

用途：把 1 分钟 K 线喂入，按 interval 分发到分钟/小时/日线窗口合成。

7.4 update_bar_minute_window(bar)

合成 N 分钟 K 线。核心逻辑：用 window_bar 累积，close_price 取最新、high/low 取 max/min、volume/turnover 累加；当 (bar.datetime.minute + 1) % window == 0 时窗口完成，调用 on_window_bar(window_bar)。

7.5 update_bar_hour_window(bar) 与 on_hour_bar(bar)

合成 N 小时 K 线。update_bar_hour_window 先合成 1 小时 K 线（以 minute == 59 或小时切换为完成标志），再交给 on_hour_bar 按 window 累积成 N 小时 K 线。

7.6 update_bar_daily_window(bar)

合成日 K 线。以 bar.datetime.time() == self.daily_end 作为日线完成标志，完成时把时间归零到当日 0 点再回调 on_window_bar。

7.7 generate() -> BarData | None

def generate(self) -> BarData | None:
    bar = self.bar
    if bar:
        bar.datetime = bar.datetime.replace(second=0, microsecond=0)
        self.on_bar(bar)
    self.bar = None
    return bar

用途：强制把当前未完成的 1 分钟 K 线立即生成并回调。一般用于收盘前强制输出最后一根 K 线。

7.8 一个容易搞错的点：多周期合成的「接力」模式

看官方 BollChannelStrategy（合成 15 分钟 K 线）：

def on_init(self):
    self.bg = BarGenerator(self.on_bar, 15, self.on_15min_bar)

def on_tick(self, tick):
    self.bg.update_tick(tick)      # tick -> 1分钟K线 -> 调 self.on_bar

def on_bar(self, bar):
    self.bg.update_bar(bar)        # 1分钟K线 -> 累积 -> 完成时调 self.on_15min_bar

def on_15min_bar(self, bar):
    # 这里才是真正的交易逻辑

数据流是：tick → 1 分钟 K 线 → 策略的 on_bar（作为中转）→ bg.update_bar → 15 分钟 K 线 → on_15min_bar。

而 MultiTimeframeStrategy 用两个 BarGenerator 合成 5 分钟 + 15 分钟：

self.bg5 = BarGenerator(self.on_bar, 5, self.on_5min_bar)
self.bg15 = BarGenerator(self.on_bar, 15, self.on_15min_bar)

def on_tick(self, tick):
    self.bg5.update_tick(tick)     # 只有 bg5 接收 tick

def on_bar(self, bar):             # bar 是 bg5 合成的 1 分钟 K 线
    self.bg5.update_bar(bar)       # 5 分钟窗口累积
    self.bg15.update_bar(bar)      # 15 分钟窗口累积（复用同一根 1 分钟 K 线）

要点：只有 bg5 直接接收 tick；bg15 通过共享的 on_bar 拿到 1 分钟 K 线后自行累积。

8. ArrayManager：技术指标计算器完全解析

来源：vnpy/trader/utility.py。ArrayManager 是一个基于 numpy + talib 的时序容器，负责存 K 线数据并计算技术指标。

8.1 构造函数与数据更新

def __init__(self, size: int = 100) -> None:
    self.count = 0
    self.size = size
    self.inited = False

    self.open_array = np.zeros(size)
    self.high_array = np.zeros(size)
    self.low_array = np.zeros(size)
    self.close_array = np.zeros(size)
    self.volume_array = np.zeros(size)
    self.turnover_array = np.zeros(size)
    self.open_interest_array = np.zeros(size)

属性	说明
`size`	内部数组长度（默认 100，即最多存 100 根 K 线，超过会滚动丢弃最旧的）
`count`	已更新的 K 线数量
`inited`	是否已攒满 `size` 根 K 线（`count >= size` 后为 True）

def update_bar(self, bar: BarData) -> None:
    self.count += 1
    if not self.inited and self.count >= self.size:
        self.inited = True
    # 数组整体左移一位，丢弃最旧值
    self.open_array[:-1] = self.open_array[1:]
    ...
    # 最新值放到数组末尾
    self.close_array[-1] = bar.close_price
    ...

用法：每根 K 线调用 self.am.update_bar(bar)，然后判断 if not self.am.inited: return（数据不足时跳过交易逻辑）。

8.2 数据访问属性（property）

属性	返回	含义
`self.am.open`	`np.ndarray`	开盘价序列
`self.am.high`	`np.ndarray`	最高价序列
`self.am.low`	`np.ndarray`	最低价序列
`self.am.close`	`np.ndarray`	收盘价序列
`self.am.volume`	`np.ndarray`	成交量序列
`self.am.turnover`	`np.ndarray`	成交额序列
`self.am.open_interest`	`np.ndarray`	持仓量序列

这些属性对应内部数组，可配合 talib 或自定义 numpy 计算直接使用。

8.3 指标函数通用约定

ArrayManager 的每个指标函数都遵循同一套 array 参数约定：

array=False（默认）：返回最新一个值（float 或元组）。
array=True：返回整个结果数组（np.ndarray 或元组）。

例如 self.am.sma(10) 返回最新 SMA 值；self.am.sma(10, array=True) 返回整条 SMA 数组（可用来取 [-1]、[-2] 判断金叉）。

8.4 单值指标函数一览

下面函数签名均摘自源码，括号内为「TA-Lib 名称 / 说明」。

均线类

方法	签名	说明
`sma`	`sma(n, array=False)`	简单移动平均
`ema`	`ema(n, array=False)`	指数移动平均
`kama`	`kama(n, array=False)`	考夫曼自适应均线
`wma`	`wma(n, array=False)`	加权移动平均

动量类

方法	签名	说明
`apo`	`apo(fast_period, slow_period, matype=0, array=False)`	绝对价格振荡
`cmo`	`cmo(n, array=False)`	钱德动量摆动
`mom`	`mom(n, array=False)`	动量
`ppo`	`ppo(fast_period, slow_period, matype=0, array=False)`	百分比价格振荡
`roc`	`roc(n, array=False)`	变动率
`rocr`	`rocr(n, array=False)`	变动率比值
`rocp`	`rocp(n, array=False)`	变动率百分比
`rocr_100`	`rocr_100(n, array=False)`	变动率比值×100
`trix`	`trix(n, array=False)`	三重指数平滑均线

统计/波动类

方法	签名	说明
`std`	`std(n, nbdev=1, array=False)`	标准差
`obv`	`obv(array=False)`	能量潮
`cci`	`cci(n, array=False)`	顺势指标
`atr`	`atr(n, array=False)`	平均真实波幅
`natr`	`natr(n, array=False)`	归一化真实波幅
`rsi`	`rsi(n, array=False)`	相对强弱指标
`trange`	`trange(array=False)`	真实波幅

趋势类（方向性指标）

方法	签名	说明
`adx`	`adx(n, array=False)`	平均趋向指数
`adxr`	`adxr(n, array=False)`	平均趋向指数评级
`dx`	`dx(n, array=False)`	趋向指数
`minus_di`	`minus_di(n, array=False)`	负向指标
`plus_di`	`plus_di(n, array=False)`	正向指标
`minus_dm`	`minus_dm(n, array=False)`	负向动向
`plus_dm`	`plus_dm(n, array=False)`	正向动向

其他

方法	签名	说明
`willr`	`willr(n, array=False)`	威廉指标
`ultosc`	`ultosc(tp1=7, tp2=14, tp3=28, array=False)`	终极振荡器
`mfi`	`mfi(n, array=False)`	资金流量指标
`ad`	`ad(array=False)`	佳庆指标
`adosc`	`adosc(fast_period, slow_period, array=False)`	佳庆摆动指标
`bop`	`bop(array=False)`	均势指标
`aroonosc`	`aroonosc(n, array=False)`	阿隆振荡
`sar`	`sar(acceleration, maximum, array=False)`	抛物线转向

8.5 多返回值指标函数

这几个函数返回元组（array=False 时返回两个/三个最新值，array=True 时返回两个/三个数组）：

方法	签名	返回值
`macd`	`macd(fast_period, slow_period, signal_period, array=False)`	`(macd, signal, hist)`
`boll`	`boll(n, dev, array=False)`	`(up, down)` 布林上下轨
`keltner`	`keltner(n, dev, array=False)`	`(up, down)` 肯特纳上下轨
`donchian`	`donchian(n, array=False)`	`(up, down)` 唐奇安上下轨
`aroon`	`aroon(n, array=False)`	`(aroon_up, aroon_down)`
`stoch`	`stoch(fastk_period, slowk_period, slowk_matype, slowd_period, slowd_matype, array=False)`	`(k, d)`

用法示例：

# 布林带（返回两个最新值）
self.boll_up, self.boll_down = self.am.boll(self.boll_window, self.boll_dev)

# MACD（返回三个最新值）
macd, signal, hist = self.am.macd(12, 26, 9)

# 唐奇安通道（海龟策略入场/出场通道）
self.entry_up, self.entry_down = self.am.donchian(self.entry_window)

# SMA 取整条数组，比较 [-1] 与 [-2] 判断金叉
fast_ma = self.am.sma(self.fast_window, array=True)
self.fast_ma0 = fast_ma[-1]
self.fast_ma1 = fast_ma[-2]

8.6 ArrayManager 底层依赖说明

ArrayManager 内部全部调用 talib（import talib）。因此如果你需要 ArrayManager 未封装的指标，可以直接 talib.XXX(self.am.high, self.am.low, self.am.close, ...) 自行计算。

9. CtaSignal 与 TargetPosTemplate

9.1 CtaSignal（信号基类）

来源：template.py。CtaSignal 用于把「信号生成」与「下单执行」解耦——每个信号只负责输出一个目标仓位（signal_pos），由上层策略汇总后统一交易。

class CtaSignal(ABC):
    def __init__(self) -> None:
        self.signal_pos = 0

    def on_tick(self, tick: TickData) -> None: ...
    @abstractmethod
    def on_bar(self, bar: BarData) -> None: ...

    def set_signal_pos(self, pos: int) -> None:
        self.signal_pos = pos

    def get_signal_pos(self) -> Any:
        return self.signal_pos

成员	说明
`signal_pos`	该信号输出的目标仓位（-1 空 / 0 平 / 1 多，或更大数值表示叠加）
`on_tick(tick)`	tick 回调（默认空实现）
`on_bar(bar)`	bar 回调，抽象方法，子类必须实现
`set_signal_pos(pos)`	设置目标仓位
`get_signal_pos()`	读取目标仓位

信号类需要自己实例化 BarGenerator / ArrayManager，如官方 RsiSignal：

class RsiSignal(CtaSignal):
    def __init__(self, rsi_window: int, rsi_level: float) -> None:
        super().__init__()
        self.rsi_window = rsi_window
        self.rsi_level = rsi_level
        self.rsi_long = 50 + self.rsi_level
        self.rsi_short = 50 - self.rsi_level
        self.bg = BarGenerator(self.on_bar)
        self.am = ArrayManager()

    def on_tick(self, tick):
        self.bg.update_tick(tick)

    def on_bar(self, bar):
        self.am.update_bar(bar)
        if not self.am.inited:
            self.set_signal_pos(0)
        rsi_value = self.am.rsi(self.rsi_window)
        if rsi_value >= self.rsi_long:
            self.set_signal_pos(1)
        elif rsi_value <= self.rsi_short:
            self.set_signal_pos(-1)
        else:
            self.set_signal_pos(0)

9.2 TargetPosTemplate（目标仓位模板）

来源：template.py。它继承自 CtaTemplate，提供「声明一个目标仓位 target_pos，模板自动完成追价下单与撤单重挂」的机制。适用于多信号叠加等「我只关心最终要多少仓位」的场景。

class TargetPosTemplate(CtaTemplate):
    tick_add = 1

    last_tick: TickData | None = None
    last_bar: BarData | None = None
    target_pos = 0

    def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
        super().__init__(cta_engine, strategy_name, vt_symbol, setting)
        self.active_orderids: list[str] = []
        self.cancel_orderids: list[str] = []
        self.variables.append("target_pos")

成员	说明
`tick_add`	追价偏移量（默认 `1`，下单时直接加到对手价上，如 `ask_price_1 + tick_add`；具体是否等于一个最小变动价位取决于合约 pricetick）
`last_tick` / `last_bar`	缓存最新 tick / bar，用于下单时确定价格
`target_pos`	目标仓位（会被自动追加到 `variables`）
`active_orderids`	当前活跃委托 id 列表
`cancel_orderids`	正在撤销的委托 id 列表

核心方法：

set_target_pos(target_pos)：设置目标仓位并立即触发 trade()。
trade()：若还有未完成委托则 cancel_old_order()，否则 send_new_order()。
cancel_old_order()：撤销所有活跃委托。
send_new_order()：根据 target_pos - self.pos 的差值计算需要增减的仓位，按 tick 或 bar 价格追价下单（回测里直接 buy/short，实盘里区分平仓/开仓逻辑，并处理涨跌停价限制）。
check_order_finished()：判断是否所有委托都已了结。

用法：继承 TargetPosTemplate，在逻辑里调用 self.set_target_pos(目标仓位)，其余交给模板。官方 MultiSignalStrategy 就是这种模式（三个信号各自输出 signal_pos，求和后 set_target_pos）。

10. 官方示例策略逐行解读

strategies/ 目录下有 9 个官方示例，涵盖了几乎所有常用写法。逐一说明它们演示了什么：

文件	演示要点
`double_ma_strategy.py`	双均线金叉死叉，`sma(array=True)` 取 `[-1]/[-2]` 判断交叉
`atr_rsi_strategy.py`	ATR + RSI 入场，`stop=True` 移动止损
`boll_channel_strategy.py`	合成 15 分钟 K 线 + 布林带 + CCI + ATR 止损
`dual_thrust_strategy.py`	日内突破（Dual Thrust），用 `bar.datetime.date()` 判断跨日、`time()` 判断收盘平仓
`king_keltner_strategy.py`	合成 5 分钟 K 线 + 肯特纳通道，自定义 `send_oco_order`（OCO 二选一单）
`multi_signal_strategy.py`	`CtaSignal` 多信号 + `TargetPosTemplate` 目标仓位
`multi_timeframe_strategy.py`	双周期（5 分钟 + 15 分钟）`BarGenerator`
`turtle_signal_strategy.py`	海龟交易法，唐奇安通道 + ATR 加仓/止损
`test_strategy.py`	下单接口自测（市价/限价/停止单/撤单）

10.1 double_ma_strategy.py 完整解读

这是最基础的模板，建议先吃透它：

import numpy as np
from vnpy_ctastrategy import (
    CtaTemplate, StopOrder, TickData, BarData,
    TradeData, OrderData, BarGenerator, ArrayManager,
)

class DoubleMaStrategy(CtaTemplate):
    author = "用Python的交易员"

    fast_window: int = 10      # 参数：快线周期
    slow_window: int = 20      # 参数：慢线周期

    fast_ma0: float = 0.0      # 变量：快线当前值
    fast_ma1: float = 0.0      # 变量：快线前一根值
    slow_ma0: float = 0.0
    slow_ma1: float = 0.0

    parameters = ["fast_window", "slow_window"]
    variables = ["fast_ma0", "fast_ma1", "slow_ma0", "slow_ma1"]

    def on_init(self):
        self.write_log("策略初始化")
        self.bg = BarGenerator(self.on_bar)
        self.am = ArrayManager()
        self.load_bar(10)          # 预热 10 天

    def on_tick(self, tick):
        self.bg.update_tick(tick)  # tick 合成 1 分钟 K 线

    def on_bar(self, bar):
        self.cancel_all()          # 每根新 K 线先撤未成交单
        am = self.am
        am.update_bar(bar)
        if not am.inited:
            return

        # 取整条均线数组，[-1] 当前值 [-2] 前一根值
        fast_ma = am.sma(self.fast_window, array=True)
        self.fast_ma0 = fast_ma[-1]
        self.fast_ma1 = fast_ma[-2]
        slow_ma = am.sma(self.slow_window, array=True)
        self.slow_ma0 = slow_ma[-1]
        self.slow_ma1 = slow_ma[-2]

        # 金叉 / 死叉判定
        cross_over = self.fast_ma0 > self.slow_ma0 and self.fast_ma1 < self.slow_ma1
        cross_below = self.fast_ma0 < self.slow_ma0 and self.fast_ma1 > self.slow_ma1

        if cross_over:
            if self.pos == 0:
                self.buy(bar.close_price, 1)
            elif self.pos < 0:
                self.cover(bar.close_price, 1)   # 先平空
                self.buy(bar.close_price, 1)     # 再开多
        elif cross_below:
            if self.pos == 0:
                self.short(bar.close_price, 1)
            elif self.pos > 0:
                self.sell(bar.close_price, 1)    # 先平多
                self.short(bar.close_price, 1)   # 再开空

        self.put_event()          # 刷新界面

    def on_order(self, order): pass
    def on_trade(self, trade): self.put_event()
    def on_stop_order(self, stop_order): pass

注意点：

self.cancel_all() 放在 on_bar 开头，确保上一根 K 线没成交的挂单作废，避免「陈旧委托」在下一根 K 线成交。
反手时用「先平后开」：cover + buy 或 sell + short。

11. 回测引擎 BacktestingEngine

来源：vnpy_ctastrategy/backtesting.py。回测引擎与 CtaTemplate 通过同一套接口交互（send_order/cancel_order/load_bar/get_pricetick 等），因此同一个策略类无需修改即可在回测与实盘间切换。

11.1 常用回测方法

方法	作用
`set_parameters(vt_symbol, interval, start, rate, slippage, size, pricetick, capital=0, end=None, mode=BAR, risk_free=0, annual_days=240, half_life=120)`	设置回测参数
`add_strategy(strategy_class, setting)`	绑定策略类与参数
`load_data()`	加载历史数据
`run_backtesting()`	执行回测
`calculate_result()`	计算逐日盯市盈亏，返回 `DataFrame`
`calculate_statistics(df=None, output=True)`	计算绩效统计指标，返回 `dict`
`show_chart(df=None)`	生成 plotly 图表（资金曲线/回撤/日盈亏/分布）
`run_bf_optimization(optimization_setting, output=True, max_workers=None)`	穷举参数优化
`run_ga_optimization(optimization_setting, ...)`	遗传算法参数优化
`get_all_trades()` / `get_all_orders()` / `get_all_daily_results()`	取回测成交/委托/逐日结果

11.2 set_parameters 参数说明

参数	含义
`vt_symbol`	合约代码（`"rb2410.SHFE"`）
`interval`	回测 K 线周期
`start`	回测开始时间
`end`	回测结束时间（缺省为当前时间）
`rate`	手续费率
`slippage`	滑点（每手）
`size`	合约乘数
`pricetick`	最小变动价位
`capital`	起始资金（默认 100 万）
`mode`	`BacktestingMode.BAR`（K 线模式，默认）或 `BacktestingMode.TICK`（tick 模式）
`risk_free`	无风险利率（用于夏普）
`annual_days`	年化交易日数（默认 240）
`half_life`	EWM 夏普的半衰期（默认 120）

11.3 回测成交撮合规则（重要，影响策略表现）

回测在每根 K 线（或每个 tick）到来时，先撮合挂单再调用策略回调。具体规则见 cross_limit_order 与 cross_stop_order：

限价单（cross_limit_order）——K 线模式下：

多头成交条件：order.price >= bar.low_price（价格触及）
空头成交条件：order.price <= bar.high_price
成交价：多头 min(order.price, bar.open_price)；空头 max(order.price, bar.open_price)（即「保守成交价」，用开盘价封顶/兜底）

停止单（cross_stop_order）——K 线模式下：

多头触发条件：stop_order.price <= bar.high_price
空头触发条件：stop_order.price >= bar.low_price
成交价：多头 max(stop_order.price, bar.open_price)；空头 min(stop_order.price, bar.open_price)

这些规则意味着：同一根 K 线内既触发又成交，成交价有保守处理，与实盘的「五档对手价」逻辑存在差异——这是回测与实盘产生偏差的常见来源，理解它有助于正确解释回测结果。

11.4 回测中策略可用的引擎接口

BacktestingEngine 实现了与 CtaEngine 相同的接口，因此策略里调用 self.get_engine_type()、self.get_pricetick()、self.get_size()、self.load_bar()、self.buy() 等都能正常工作。区别在于：

send_notification / send_email：回测中为空实现（pass），不会真的发通知。
sync_strategy_data / put_strategy_event：回测中为空实现（pass）。
get_pricetick/get_size 返回 set_parameters 传入的 pricetick/size。

12. 实盘引擎 CtaEngine 与策略生命周期

来源：vnpy_ctastrategy/engine.py。理解引擎如何驱动策略，有助于你写出正确、健壮的策略。

12.1 引擎启动流程

def init_engine(self):
    self.init_datafeed()          # 初始化数据服务
    self.load_strategy_class()    # 加载策略类（扫描 strategies 目录）
    self.load_strategy_setting()  # 从 cta_strategy_setting.json 恢复策略实例
    self.load_strategy_data()     # 从 cta_strategy_data.json 恢复策略状态
    self.register_event()         # 注册 tick/order/trade 事件

12.2 事件驱动流程

引擎订阅了三个主事件，并转发给对应策略：

EVENT_TICK → process_tick_event：先检查本地停止单是否触发（check_stop_order），再对每个 inited 的策略调用 on_tick(tick)。
EVENT_ORDER → process_order_event：根据 vt_orderid 找到所属策略，调用 on_order(order)；若委托是 OrderType.STOP，还会额外构造 StopOrder 调用 on_stop_order。
EVENT_TRADE → process_trade_event：去重后找到所属策略，先更新 strategy.pos，再调用 on_trade(trade)，随后自动同步数据到磁盘并刷新界面。

关键点：process_trade_event 里持仓更新逻辑是：

if trade.direction == Direction.LONG:
    strategy.pos += trade.volume
else:
    strategy.pos -= trade.volume

即 LONG 加仓、SHORT 减仓，pos 正负分别代表净多头/净空头。

12.3 策略实例的生命周期

创建（add_strategy）：校验重名、类存在、vt_symbol 格式合法后实例化，写入 strategies 与 symbol_strategy_map。
初始化（_init_strategy，在单独线程执行）：
- 调用 strategy.on_init()
- 若 strategy_data 里有历史记录，则回写 variables（这是变量持久化恢复的关键）
- 订阅行情
- 置 strategy.inited = True
启动（start_strategy）：要求已 inited；调用 on_start()；置 trading = True。
停止（stop_strategy）：调用 on_stop()；置 trading = False；cancel_all(strategy) 撤掉所有未成交委托；同步数据。
移除（remove_strategy）：要求已停止；清理设置、映射关系、从字典删除。

12.4 停止单的本地/服务器分流

send_order 中，当 stop=True 时：

if stop:
    if contract.stop_supported:
        return self.send_server_stop_order(...)   # 服务器支持则发服务器停止单
    else:
        return self.send_local_stop_order(...)    # 否则用本地停止单
else:
    return self.send_limit_order(...)

服务器停止单：直接以 OrderType.STOP 下单，由交易所/柜台托管。
本地停止单：由引擎在内存维护（stop_orders），每次 tick 到来时在 check_stop_order 里判断触发，触发后以对手五档价发限价单。本地停止单的 stop_orderid 以 "STOP." 开头（STOPORDER_PREFIX = "STOP"）。

下单前，引擎还会做价格与数量对齐：

price = round_to(price, contract.pricetick)
volume = round_to(volume, contract.min_volume)

所以策略里传的价格/数量即使不是最小变动价位的整数倍，也会被自动对齐。

12.5 异常保护

call_strategy_func 会捕获策略回调里的所有异常：

def call_strategy_func(self, strategy, func, params=None):
    try:
        if params: func(params)
        else: func()
    except Exception:
        strategy.trading = False
        strategy.inited = False
        msg = f"触发异常已停止\n{traceback.format_exc()}"
        self.write_log(msg, strategy)

含义：策略回调抛异常时，引擎会自动停止该策略（trading=False, inited=False）并打印完整堆栈到日志。所以你的策略回调里务必做好数据边界判断（如指标未就绪就 return），避免因 NaN、空数组等触发异常导致策略被意外停止。

13. 编写策略的完整步骤清单

确定文件位置：在 vnpy 启动目录下建 strategies/ 文件夹，新建 你的策略.py。
写导入：从 vnpy_ctastrategy 导入 CtaTemplate, StopOrder, TickData, BarData, TradeData, OrderData, BarGenerator, ArrayManager（用到 Direction 时再从 vnpy_ctastrategy 导入，它已在 __init__.py 中导出）。
定义类：class MyStrategy(CtaTemplate):。
写 author。
定义参数与变量：参数用类型标注 + 默认值 + 列入 parameters；内部状态列入 variables。
实现 on_init：创建 BarGenerator/ArrayManager，计算派生参数，调用 self.load_bar(10) 预热。
实现 on_tick：self.bg.update_tick(tick)。
实现 on_bar：self.am.update_bar(bar) → if not self.am.inited: return → 计算指标 → 判断信号 → 下单 → self.put_event()。
按需实现 on_trade / on_order / on_stop_order（空实现也要显式写 pass 更清晰）。
重启 vnpy：让引擎重新 load_strategy_class 加载新策略。
在 CTA策略界面创建实例：填策略名、vt_symbol、参数，点「初始化」→「启动」。
先回测验证：用回测引擎跑一遍，确认逻辑与指标值符合预期，再上实盘。

14. 常见陷阱与最佳实践

指标未就绪就下单：ArrayManager 需要攒满 size 根 K 线才 inited。务必在 on_bar 里 if not self.am.inited: return，否则 talib 会返回全 NaN，导致逻辑错误甚至异常（异常会让引擎自动停止策略）。
pos 是引擎自动维护的，不要手动改。反手时用「先平后开」（cover + buy / sell + short）。
每根新 K 线先 cancel_all()：避免上一根 K 线挂的限价/停止单在下一根 K 线「迟到成交」，造成计划外仓位。
self.put_event() 别忘：不加它，界面上 variables（持仓、指标）不会实时刷新，但不影响实际交易。
异常即停：任何回调抛异常都会被引擎捕获并停止策略。对 NaN、除零、空 tick（last_price == 0）等边界情况做好防护。
回测 ≠ 实盘：
- 回测停止单成交价用 open_price 保守撮合，实盘用对手五档价；
- 回测 send_notification/sync_data/put_event 都是空操作；
- 回测手续费/滑点靠 set_parameters 传入的 rate/slippage 估算。
lock 与 net 参数：这两个参数用于适配不同交易所的持仓/平仓规则（锁仓模式、净持仓模式），由 main_engine.convert_order_request 统一转换下单请求。普通单边策略保持默认 False 即可。
多周期合成注意「接力」：只有「直接接收 tick 的那个 BarGenerator」需要 update_tick；其余 BarGenerator 通过共享的 on_bar 用 update_bar 驱动（见 7.8 节）。
N 分钟窗口限制：合成 N 分钟 K 线时 N 必须能整除 60（2/3/5/6/10/15/20/30），否则分钟窗口无法正确对齐。
vt_symbol 必须带交易所后缀：形如 "rb2410.SHFE"，否则创建策略实例会被引擎拒绝（add_strategy 中有格式校验）。
停止单的 OCO 用法：开仓时想「突破上轨买、跌破下轨卖」可用两个 stop=True 的单子（参考 king_keltner_strategy.send_oco_order），成交一腿后手动撤另一腿。

附录：核心源码文件速查表

想查的内容	文件	关键类/函数
策略基类全部方法	`vnpy_ctastrategy/template.py`	`CtaTemplate` / `CtaSignal` / `TargetPosTemplate`
停止单对象	`vnpy_ctastrategy/base.py`	`StopOrder` / `StopOrderStatus` / `EngineType` / `BacktestingMode`
实盘引擎	`vnpy_ctastrategy/engine.py`	`CtaEngine`
回测引擎	`vnpy_ctastrategy/backtesting.py`	`BacktestingEngine`
数据结构	`vnpy/trader/object.py`	`BarData` / `TickData` / `OrderData` / `TradeData` / `ContractData`
枚举常量	`vnpy/trader/constant.py`	`Direction` / `Offset` / `Interval` / `Exchange` / `Status` / `OrderType`
K 线合成与指标	`vnpy/trader/utility.py`	`BarGenerator` / `ArrayManager` / `round_to` / `extract_vt_symbol`
官方示例	`vnpy_ctastrategy/strategies/*.py`	9 个示例策略

无意

Member