Dual Thrust Breakout Entries from the Prior Day’s Range
Summary
This code implements a daily range breakout strategy in a CTA framework. At the start of a new day, it uses the preceding day’s high-minus-low range to set an upper entry at the new day’s open plus 0.4 times that range, and a lower entry at the open minus 0.6 times the range. Before 14:55, it chooses a direction based on whether the current bar closes above or below the day’s open, then submits stop orders at the corresponding levels. The position logic can reverse between long and short as the opposing threshold is reached.
The code tracks intraday highs and lows, resets entry flags each day, and attempts to close open positions after the stated exit time. It specifies a fixed position size of one, but supplies no backtest results or evidence of profitability. The implementation depends on bar timing, order handling, and the framework’s stop-order behavior; its exit and reversal mechanics therefore need careful evaluation before use. The source includes no explicit loss-based risk sizing or volatility filter.
Key ideas
- The strategy sets daily breakout thresholds using the prior day’s range and the current day’s opening price.
- The upper threshold uses 0.4 times the range, while the lower threshold uses 0.6 times the range.
- The close relative to the day’s open selects an initial long or short direction before the scheduled exit time.
- Stop orders at the opposing threshold support position reversal, and open positions are handled after 14:55.
- The code provides no performance evidence and uses a fixed position size of one.
Tags
Full text
# DualThrustStrategy
# DualThrustStrategy
## Source (MIT)
```python
from datetime import time
from howtrader.app.cta_strategy import (
CtaTemplate,
StopOrder
)
from howtrader.trader.object import TickData, BarData, TradeData, OrderData
from howtrader.trader.utility import BarGenerator, ArrayManager
from decimal import Decimal
class DualThrustStrategy(CtaTemplate):
""""""
author = "用Python的交易员"
fixed_size = 1
k1 = 0.4
k2 = 0.6
bars = []
day_open = 0
day_high = 0
day_low = 0
day_range = 0
long_entry = 0
short_entry = 0
exit_time = time(hour=14, minute=55)
long_entered = False
short_entered = False
parameters = ["k1", "k2", "fixed_size"]
variables = ["day_range", "long_entry", "short_entry"]
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
""""""
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
self.bg = BarGenerator(self.on_bar)
self.am = ArrayManager()
self.bars = []
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.load_bar(10)
def on_start(self):
"""
Callback when strategy is started.
"""
self.write_log("策略启动")
def on_stop(self):
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")
def on_tick(self, tick: TickData):
"""
Callback of new tick data update.
"""
self.bg.update_tick(tick)
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.cancel_all()
self.bars.append(bar)
if len(self.bars) <= 2:
return
else:
self.bars.pop(0)
last_bar = 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(Decimal(self.long_entry), Decimal(self.fixed_size), stop=True)
else:
if not self.short_entered:
self.short(Decimal(self.short_entry), Decimal(self.fixed_size), stop=True)
elif self.pos > 0:
self.long_entered = True
self.sell(Decimal(self.short_entry), Decimal(self.fixed_size), stop=True)
if not self.short_entered:
self.short(Decimal(self.short_entry), Decimal(self.fixed_size), stop=True)
elif self.pos < 0:
self.short_entered = True
self.cover(Decimal(self.long_entry), Decimal(self.fixed_size), stop=True)
if not self.long_entered:
self.buy(Decimal(self.long_entry), Decimal(self.fixed_size), stop=True)
else:
if self.pos > 0:
price = bar.close_price * 0.99
self.sell(Decimal(price), Decimal(abs(self.pos)))
elif self.pos < 0:
price = bar.close_price * 1.01
self.cover(Decimal(price), Decimal(abs(self.pos)))
self.put_event()
def on_order(self, order: OrderData):
"""
Callback of new order data update.
"""
pass
def on_trade(self, trade: TradeData):
"""
Callback of new trade data update.
"""
self.put_event()
def on_stop_order(self, stop_order: StopOrder):
"""
Callback of stop order update.
"""
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.