Double Moving Average Crossover Strategy with Position Reversals
Summary
This bar-based strategy compares a fast simple moving average with a slower one, using a crossover to choose direction. Its defaults are 10 and 20 periods. When the fast average moves above the slow average, it buys if flat or covers a short and opens a long. When the fast average moves below the slow average, it shorts if flat or sells a long and opens a short. It updates the averages on each new bar and cancels outstanding orders before evaluating the signal.
The document is a strategy implementation rather than a performance study: it reports no market, test period, transaction costs, or backtest results. It therefore offers no evidence of profitability. The crossover rule is straightforward to implement, but moving-average signals can lag and may repeatedly reverse in choppy conditions. The code also submits orders at the bar's close price; actual fills, slippage, and the handling of reversal orders depend on the trading framework and execution conditions.
Key ideas
- The default fast and slow simple moving averages use 10 and 20 periods.
- An upward crossover opens a long position or reverses a short position to long.
- A downward crossover opens a short position or reverses a long position to short.
- The implementation evaluates signals on each new bar and cancels outstanding orders first.
- No backtest results or market-specific evidence are supplied.
Tags
Full text
# DoubleMaStrategy
# DoubleMaStrategy
双均线策略。
用快慢均线金叉死叉开平仓的策略。
## Source (MIT)
```python
"""双均线策略。"""
import numpy as np
from vnpy_ctastrategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
class DoubleMaStrategy(CtaTemplate):
"""用快慢均线金叉死叉开平仓的策略。"""
author: str = "用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: list[str] = ["fast_window", "slow_window"]
variables: list[str] = ["fast_ma0", "fast_ma1", "slow_ma0", "slow_ma1"]
def on_init(self) -> None:
"""
策略初始化完成时的回调。
"""
self.write_log("策略初始化")
self.bg: BarGenerator = BarGenerator(self.on_bar)
self.am: ArrayManager = ArrayManager()
self.load_bar(10)
def on_start(self) -> None:
"""
策略启动时的回调。
"""
self.write_log("策略启动")
self.put_event()
def on_stop(self) -> None:
"""
策略停止时的回调。
"""
self.write_log("策略停止")
self.put_event()
def on_tick(self, tick: TickData) -> None:
"""
新 Tick 数据更新时的回调。
"""
self.bg.update_tick(tick)
def on_bar(self, bar: BarData) -> None:
"""
新 K 线数据更新时的回调。
"""
self.cancel_all()
am: ArrayManager = self.am
am.update_bar(bar)
if not am.inited:
return
fast_ma: np.ndarray = am.sma(self.fast_window, array=True)
self.fast_ma0 = fast_ma[-1]
self.fast_ma1 = fast_ma[-2]
slow_ma: np.ndarray = am.sma(self.slow_window, array=True)
self.slow_ma0 = slow_ma[-1]
self.slow_ma1 = slow_ma[-2]
cross_over: bool = self.fast_ma0 > self.slow_ma0 and self.fast_ma1 < self.slow_ma1
cross_below: bool = 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: OrderData) -> None:
"""
新委托数据更新时的回调。
"""
pass
def on_trade(self, trade: TradeData) -> None:
"""
新成交数据更新时的回调。
"""
self.put_event()
def on_stop_order(self, stop_order: StopOrder) -> None:
"""
停止单更新时的回调。
"""
pass
```Shown in full with attribution under the source's licence. Licence: MIT
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.