策略示例

前面几章学了零件，本章把它们组装成两个能跑的完整策略：一个双均线趋势策略（K 线驱动 + 金叉死叉 + 超价报单），一个 TWAP 拆单策略（定时器 + 分批报单 + 追单）。读懂这两个，绝大多数场景都能自己改写。

一、双均线趋势策略

策略逻辑：在可调周期的 K 线上算快慢两条 SMA，金叉开多、死叉开空，每次先平掉反向仓位再开仓，报单用超价（pay_up）保证成交。

参数与状态映射

from pythongo.base import BaseParams, BaseState, Field
from pythongo.core import KLineStyleType


class Params(BaseParams):
    """参数映射模型"""
    exchange: str = Field(default="", title="交易所代码")
    instrument_id: str = Field(default="", title="合约代码")
    fast_period: int = Field(default=5, title="短均线周期", ge=2)
    slow_period: int = Field(default=20, title="长均线周期")
    order_volume: int = Field(default=1, title="下单手数", ge=1)
    kline_style: KLineStyleType = Field(default="M1", title="K 线周期")
    pay_up: int | float = Field(default=0, title="超价")


class State(BaseState):
    """状态映射模型"""
    fast_ma: float = Field(default=0, title="短均线")
    slow_ma: float = Field(default=0, title="长均线")

策略骨架与指标计算

from pythongo.ui import BaseStrategy
from pythongo.utils import KLineGenerator
from pythongo.classdef import KLineData, TickData


class DemoDMA(BaseStrategy):
    """可调节 K 线周期的双均线交易策略"""
    def __init__(self) -> None:
        super().__init__()
        self.params_map = Params()
        self.state_map = State()
        self.kline_generator: KLineGenerator = None
        self.pre_fast_ma = 0          # 上一根快均线
        self.pre_slow_ma = 0          # 上一根慢均线
        self.order_id: set[int] = set()   # 未成交报单集合

    def on_start(self) -> None:
        self.kline_generator = KLineGenerator(
            real_time_callback=self.real_time_callback,
            callback=self.callback,
            exchange=self.params_map.exchange,
            instrument_id=self.params_map.instrument_id,
            style=self.params_map.kline_style
        )
        self.kline_generator.push_history_data()
        super().on_start()

    def on_tick(self, tick: TickData) -> None:
        super().on_tick(tick)
        self.kline_generator.tick_to_kline(tick)

    def calc_indicator(self) -> None:
        """用 array=True 取整条序列，取最后两根判断交叉"""
        slow = self.kline_generator.producer.sma(self.params_map.slow_period, array=True)
        fast = self.kline_generator.producer.sma(self.params_map.fast_period, array=True)
        self.pre_slow_ma, self.state_map.slow_ma = slow[-2:]
        self.pre_fast_ma, self.state_map.fast_ma = fast[-2:]

交叉判断与报单

    def is_cross(self) -> int:
        """返回 1 无信号 / 2 金叉 / 3 死叉"""
        if self.pre_fast_ma <= self.pre_slow_ma and self.state_map.fast_ma > self.state_map.slow_ma:
            return 2
        if self.pre_fast_ma >= self.pre_slow_ma and self.state_map.fast_ma < self.state_map.slow_ma:
            return 3
        return 1

    def callback(self, kline: KLineData) -> None:
        """每根 K 线成型触发——先撤未成交单、算指标、再按信号下单"""
        for oid in list(self.order_id):
            self.cancel_order(oid)

        self.calc_indicator()
        position = self.get_position(self.params_map.instrument_id)

        match self.is_cross():
            case 2:  # 金叉：先平空再开多
                price = kline.close + self.params_map.pay_up
                if position.net_position < 0:
                    self.order_id.add(self.auto_close_position(
                        exchange=self.params_map.exchange,
                        instrument_id=self.params_map.instrument_id,
                        volume=abs(position.net_position), price=price,
                        order_direction="buy"
                    ))
                self.order_id.add(self.send_order(
                    exchange=self.params_map.exchange,
                    instrument_id=self.params_map.instrument_id,
                    volume=self.params_map.order_volume, price=price,
                    order_direction="buy"
                ))
            case 3:  # 死叉：先平多再开空
                price = kline.close - self.params_map.pay_up
                if position.net_position > 0:
                    self.order_id.add(self.auto_close_position(
                        exchange=self.params_map.exchange,
                        instrument_id=self.params_map.instrument_id,
                        volume=position.net_position, price=price,
                        order_direction="sell"
                    ))
                self.order_id.add(self.send_order(
                    exchange=self.params_map.exchange,
                    instrument_id=self.params_map.instrument_id,
                    volume=self.params_map.order_volume, price=price,
                    order_direction="sell"
                ))

回测时直接把这个类和 Params 传给 backtesting.engine.run（见 策略结构与回测），实盘时加载到 PythonGO 窗口即可。

二、TWAP 拆单策略

大单一次性砸进去会冲击市场，TWAP 把总手数按时间均匀拆成多笔小单。这个例子展示定时器 + 报单管理 + 追单的综合用法。

核心思路

1. 把 total_volume 按 order_volume 拆成若干笔。
2. 用 Scheduler 定时器每隔 job_interval 秒发一笔。
3. 每笔用排队价挂限价单；另起一个定时任务检查价格，价格漂了就撤单重挂（追单）。

from datetime import datetime
from uuid import uuid4
from pythongo.base import BaseParams, BaseState, BaseStrategy, Field
from pythongo.utils import Scheduler
from pythongo.types import TypeOrderDIR


class Params(BaseParams):
    exchange: str = Field(default="", title="交易所代码")
    instrument_id: str = Field(default="", title="合约代码")
    total_time: int = Field(default=300, title="算法总时长(秒)")
    interval: int = Field(default=10, title="价格检查间隔(秒)")
    order_volume: int = Field(default=4, title="单笔委托数量")
    total_volume: int = Field(default=50, title="总委托手数")
    order_direction: TypeOrderDIR = Field(default="buy", title="买卖方向")

    @property
    def job_interval(self) -> int:
        """每笔间隔 = 总时长 / 笔数"""
        return self.total_time / (self.total_volume / self.order_volume)


class DemoTWAP(BaseStrategy):
    """TWAP 策略"""
    def __init__(self) -> None:
        super().__init__()
        self.params_map = Params()
        self.scheduler = Scheduler()
        self.scheduler.start()
        self.tick = None
        self.order_list = []      # 待发报单
        self.order_jobstore = []  # 已发报单

    def get_order_price(self, best_price=False) -> float:
        """取排队价；best_price=True 取对手价"""
        if not self.tick:
            return 0.0
        bid, ask = self.tick.bid_price1, self.tick.ask_price1
        if best_price:
            bid, ask = ask, bid
        return bid if self.params_map.order_direction == "buy" else ask

    def on_tick(self, tick) -> None:
        super().on_tick(tick)
        self.tick = tick

定时发单与检查

    def split_order(self) -> None:
        """拆单：按单笔数量把总手数切成列表"""
        from dataclasses import dataclass

        @dataclass
        class Order:
            memo: str
            volume: int
            remaining_volume: int = 0
            order_id: int = None
            price: float = None
            closed: bool = False

        n = self.params_map.total_volume // self.params_map.order_volume
        for _ in range(n):
            self.order_list.append(Order(memo=str(uuid4()).split("-")[0],
                                         volume=self.params_map.order_volume))

    def twap(self) -> None:
        """定时主任务：发一笔新单"""
        if not self.order_list:
            self.scheduler.remove_job("twap_job")
            self.pause_strategy()
            return
        order = self.order_list.pop(0)
        self.order_jobstore.append(order)
        order.order_id = self.send_order(
            exchange=self.params_map.exchange,
            instrument_id=self.params_map.instrument_id,
            volume=order.volume, price=self.get_order_price(),
            order_direction=self.params_map.order_direction, memo=order.memo
        )

    def check_price(self, order) -> None:
        """检查未成交单：价格漂了就撤单重挂"""
        if order.closed:
            self.scheduler.remove_job("check_price_job")
            return
        if order.price != self.get_order_price():
            self.cancel_order(order.order_id)
            order.order_id = self.send_order(
                exchange=self.params_map.exchange,
                instrument_id=self.params_map.instrument_id,
                volume=order.volume, price=self.get_order_price(),
                order_direction=self.params_map.order_direction, memo=order.memo
            )

    def on_order(self, order) -> None:
        super().on_order(order)
        job = next(o for o in self.order_jobstore if o.memo == order.memo)
        if order.status == "全部成交":
            job.closed = True
        else:
            job.remaining_volume = order.total_volume - order.traded_volume
            job.price = order.price

    def on_start(self) -> None:
        super().on_start()
        self.split_order()
        self.scheduler.add_job(self.twap, trigger="interval", id="twap_job",
                               seconds=self.params_map.job_interval,
                               next_run_time=datetime.now())

    def on_stop(self) -> None:
        super().on_stop()
        for job in ("twap_job", "check_price_job"):
            self.scheduler.remove_job(job)

TWAP 用 memo（报单备注）在回调里匹配报单状态——因为 order_id 会因撤单重挂而变化，memo 才是稳定标识。

小结

双均线策略串起了 K 线合成、指标、交叉判断、超价报单与自动平仓；TWAP 串起了定时器、拆单、追单与报单状态管理。 最后一步看 K 线图 UI 与定时器扩展：UI与扩展。