蜗牛之路

以下是基于vn.py框架实现的黑色系期货日内动量突破策略（螺纹钢/铁矿石）的完整代码，包含信号生成、风险控制和仓位管理模块：

from vnpy.app.cta_strategy import (
    CtaTemplate,
    StopOrder,
    TickData,
    BarData,
    TradeData,
    OrderData,
    BarGenerator,
    ArrayManager
)
from vnpy.trader.constant import Interval, Direction, Offset

class BlackMomentumStrategy(CtaTemplate):
    """黑色系日内动量突破策略"""
    author = "YourName"

    # 策略参数
    breakout_window = 24  # 2小时突破窗口（5分钟K线*24=2小时）
    volume_multiplier = 1.5  # 成交量放大倍数
    fixed_size = 1  # 每次交易手数
    stop_loss_pct = 0.005  # 固定止损0.5%
    trailing_atr_ratio = 3  # 动态止盈ATR倍数

    # 策略变量
    intraday_high = 0
    intraday_low = 0
    atr_value = 0
    current_tick = None

    parameters = [
        "breakout_window",
        "volume_multiplier",
        "fixed_size",
        "stop_loss_pct",
        "trailing_atr_ratio"
    ]
    variables = [
        "intraday_high",
        "intraday_low",
        "atr_value"
    ]

    def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
        super().__init__(cta_engine, strategy_name, vt_symbol, setting)

        # 创建K线合成器（5分钟）
        self.bg = BarGenerator(self.on_bar, interval=Interval.MINUTE, window=5)
        # 使用ArrayManager管理K线序列（默认100根）
        self.am = ArrayManager(size=100)

    def on_init(self):
        """策略初始化"""
        self.write_log("策略初始化")
        self.load_bar(10)  # 加载10天历史数据

    def on_start(self):
        """策略启动"""
        self.write_log("策略启动")
        self.put_event()

    def on_stop(self):
        """策略停止"""
        self.write_log("策略停止")
        self.put_event()

    def on_tick(self, tick: TickData):
        """Tick更新"""
        self.current_tick = tick
        self.bg.update_tick(tick)

    def on_bar(self, bar: BarData):
        """5分钟K线处理"""
        # 更新K线到ArrayManager
        self.am.update_bar(bar)
        if not self.am.inited:
            return

        # 计算技术指标
        self.atr_value = self.am.atr(14)
        self.intraday_high = self.am.high[-self.breakout_window:].max()
        self.intraday_low = self.am.low[-self.breakout_window:].min()
        vol_ma20 = self.am.sma(self.am.volume, 20)

        # 过滤交易时间段（上午10:00-11:30，下午13:30-15:00）
        if not (100000 <= bar.datetime.time().hour * 100 + bar.datetime.time().minute <= 113000 or
                133000 <= bar.datetime.time().hour * 100 + bar.datetime.time().minute <= 150000):
            return

        # 撤销未成交订单
        self.cancel_all()

        # 无持仓时检查信号
        if self.pos == 0:
            # 突破上轨且放量
            if bar.close_price > self.intraday_high and bar.volume > vol_ma20 * self.volume_multiplier:
                self.buy(bar.close_price + 5, self.fixed_size)
            # 突破下轨且放量
            elif bar.close_price < self.intraday_low and bar.volume > vol_ma20 * self.volume_multiplier:
                self.short(bar.close_price - 5, self.fixed_size)

        # 有持仓时处理止损止盈
        elif self.pos > 0:  # 多头持仓
            # 固定止损
            stop_price = bar.close_price * (1 - self.stop_loss_pct)
            # 动态止盈（最高价回撤3倍ATR）
            trail_price = self.am.high[-self.breakout_window:] - self.atr_value * self.trailing_atr_ratio
            stop_price = max(stop_price, trail_price)
            self.sell(stop_price, abs(self.pos), stop=True)

        elif self.pos < 0:  # 空头持仓
            # 固定止损
            stop_price = bar.close_price * (1 + self.stop_loss_pct)
            # 动态止盈（最低价反弹3倍ATR）
            trail_price = self.am.low[-self.breakout_window:] + self.atr_value * self.trailing_atr_ratio
            stop_price = min(stop_price, trail_price)
            self.cover(stop_price, abs(self.pos), stop=True)

        self.put_event()

    def on_order(self, order: OrderData):
        """委托更新"""
        pass

    def on_trade(self, trade: TradeData):
        """成交更新"""
        self.put_event()

    def on_stop_order(self, stop_order: StopOrder):
        """停止单更新"""
        pass