Dual Thrust Breakout Strategy Using the Prior Day’s Range
Summary
This document presents a Dual Thrust intraday breakout approach. At the start of a new trading day, it calculates the prior day’s high-low range and uses the new day’s opening price to set separate upper and lower entry thresholds. The k1 and k2 parameters scale the range differently for long and short entries. Stop orders are placed during the session, with direction influenced by whether the current close is above or below the day’s open.
The implementation tracks daily highs and lows, limits new entries to before a specified exit time, and attempts to close open positions near the session end. It also describes reversing or managing positions when already long or short. The document supplies Python code for a trading framework but no market, test period, costs, or performance results. Its behavior depends on bar timing, stop-order handling, and framework conventions; these details should be checked, and the range breakout logic should be evaluated with realistic execution assumptions.
Key ideas
- The strategy derives daily breakout thresholds from the previous day’s range and current day’s open.
- Separate scaling parameters control the long and short thresholds.
- Entry logic uses stop orders and the close relative to the day’s open to select a direction.
- The implementation restricts trading near the session end and attempts to close remaining positions.
- No backtest evidence or execution-cost analysis is supplied.
Tags
Full text
# DualThrustStrategy
# DualThrustStrategy
Dual Thrust策略。
按前一日振幅和开盘价计算上下轨，盘中突破后开仓的策略。
## Source (MIT)
```python
"""Dual Thrust策略。"""
from datetime import time
from vnpy_ctastrategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
class DualThrustStrategy(CtaTemplate):
"""按前一日振幅和开盘价计算上下轨，盘中突破后开仓的策略。"""
author: str = "用Python的交易员"
fixed_size: int = 1
k1: float = 0.4
k2: float = 0.6
day_open: float = 0
day_high: float = 0
day_low: float = 0
day_range: float = 0
long_entry: float = 0
short_entry: float = 0
long_entered: bool = False
short_entered: bool = False
parameters: list[str] = ["k1", "k2", "fixed_size"]
variables: list[str] = ["day_range", "long_entry", "short_entry"]
def on_init(self) -> None:
"""
策略初始化完成时的回调。
"""
self.write_log("策略初始化")
self.bg: BarGenerator = BarGenerator(self.on_bar)
self.am: ArrayManager = ArrayManager()
self.bars: list[BarData] = []
self.exit_time: time = time(hour=14, minute=55)
self.load_bar(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)
def on_bar(self, bar: BarData) -> None:
"""
新 K 线数据更新时的回调。
"""
self.cancel_all()
self.bars.append(bar)
if len(self.bars) <= 2:
return
else:
self.bars.pop(0)
last_bar: BarData = self.bars[-2]
if last_bar.datetime.date() != bar.datetime.date():
if self.day_high:
self.day_range = self.day_high - self.day_low
self.long_entry = bar.open_price + self.k1 * self.day_range
self.short_entry = bar.open_price - self.k2 * self.day_range
self.day_open = bar.open_price
self.day_high = bar.high_price
self.day_low = bar.low_price
self.long_entered = False
self.short_entered = False
else:
self.day_high = max(self.day_high, bar.high_price)
self.day_low = min(self.day_low, bar.low_price)
if not self.day_range:
return
if bar.datetime.time() < self.exit_time:
if self.pos == 0:
if bar.close_price > self.day_open:
if not self.long_entered:
self.buy(self.long_entry, self.fixed_size, stop=True)
else:
if not self.short_entered:
self.short(self.short_entry,
self.fixed_size, stop=True)
elif self.pos > 0:
self.long_entered = True
self.sell(self.short_entry, self.fixed_size, stop=True)
if not self.short_entered:
self.short(self.short_entry, self.fixed_size, stop=True)
elif self.pos < 0:
self.short_entered = True
self.cover(self.long_entry, self.fixed_size, stop=True)
if not self.long_entered:
self.buy(self.long_entry, self.fixed_size, stop=True)
else:
if self.pos > 0:
self.sell(bar.close_price * 0.99, abs(self.pos))
elif self.pos < 0:
self.cover(bar.close_price * 1.01, abs(self.pos))
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.