VeighNa量化社区
你的开源社区量化交易平台
Member
avatar
加入于:
帖子: 5
声望: 0

程序员:黑驴

开发时间:2023/2/8 9:14

coding:utf-8

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

class Ma1cdStrategy(CtaTemplate):

author = "黑驴"

fast_window = 6
slow_window = 10
signal_window = 5
fixed_size = 1

fast_ma = 0.0
slow_ma = 0.0
ma1cd = 0.0
signal = 0.0
histogram = 0.0


parameters: list = [
    "fast_window",
    "slow_window",
    "signal_window",
    "fixed_size"
]
variables: list = [
    "fast_ma",
    "slow_ma",
    "ma1cd",
    "signal",
    "histogram"
]


def __init__(
        self,
        cta_engine: Any,
        strategy_name: str,
        vt_symbol: str,
        setting: dict,
) -> None:
    """"""
    super().__init__(cta_engine, strategy_name, vt_symbol, setting)

    self.bg15 = BarGenerator(self.on_bar, 15, self.on_15min_bar)

    self.am15 = ArrayManager()


def on_init(self) -> None:
    """
    Callback when strategy is inited.
    """
    self.write_log("策略初始化")

    self.load_bar(10)


def on_start(self) -> None:
    """
    Callback when strategy is started.
    """
    self.write_log("策略启动")


def on_stop(self) -> None:
    """
    Callback when strategy is stopped.
    """
    self.write_log("策略停止")


def on_tick(self, tick: TickData) -> None:
    """
    Callback of new tick data update.
    """
    self.bg15.update_tick(tick)


def on_bar(self, bar: BarData) -> None:
    """
    Callback of new bar data update.
    """
    self.bg15.update_bar(bar)


def on_15min_bar(self, bar: BarData):
    """"""
    self.cancel_all()

    self.am15.update_bar(bar)
    if self.am15.inited:
        return

    self.fast_ma = self.am15.sma(bar.close, self.fast_window)
    self.slow_ma = self.am15.sma(bar.close, self.slow_window)
    self.ma1cd = self.fast_ma - self.slow_ma
    self.signal = self.am15.ema(self.ma1cd, self.signal_window)
    self.histogram = self.ma1cd - self.signal

    cross_above = self.histogram > 0 and self.am15.cross_above(self.histogram, 0)
    cross_below = self.histogram < 0 and self.am15.cross_below(self.histogram, 0)

    if cross_above:
        if self.pos == 0:
            self.buy(bar.close, self.fixed_size, stop=True)
        elif self.pos < 0:
            self.cover(bar.close, abs(self.pos), stop=True)
            self.buy(bar.close, self.fixed_size, stop=True)

    elif cross_below:
        if self.pos == 0:
            self.short(bar.close, self.fixed_size, stop=True)
        elif self.pos > 0:
            self.sell(bar.close, self.pos, stop=True)
            self.short(bar.close, self.fixed_size, stop=True)

    self.put_event()



def on_trade(self, trade: TradeData) -> None:
    """
    Callback of new trade data update.
    """
    self.put_event()

def on_order(self, order: OrderData) -> None:
    """
    Callback of new order data update.
    """
    pass

def on_stop_order(self, stop_order: StopOrder) -> None:
    """
    Callback of stop order update.
    """
    pass
Member
avatar
加入于:
帖子: 4749
声望: 288

可以在on_15min_bar函数里打印一下,看看是不是历史数据太少导致加载的历史数据都被if self.am15.inited过滤了

© 2015-2022 上海韦纳软件科技有限公司
备案服务号:沪ICP备18006526号

沪公网安备 31011502017034号

【用户协议】
【隐私政策】
【免责条款】