Time-Gated Spread Trading with Position-Based Entry and Exit Algos
Summary
This example describes a spread-trading strategy that manages long and short spread positions through separate opening and closing algorithms. During a configured session, it starts both long-entry and short-entry algorithms when flat, using target prices, a maximum position, a price adjustment allowance, and an interval. Once a position exists, it stops opening algorithms and starts the matching exit algorithm for the full spread position.
The strategy checks the spread update time against its allowed window and stops all active algorithms outside that window. It also refreshes position and algorithm state through callbacks, clearing algorithm identifiers when an algo becomes inactive. The document provides implementation logic rather than performance evidence or a tested trading rationale. It does not explain how entry and exit prices should be selected, how spread risk should be sized, or how the legs are executed; those depend on the trading framework and the chosen spread.
Key ideas
- The strategy only runs its spread logic within a configured time window.
- When flat, it starts separate algorithms for long and short spread entries.
- For an existing long or short spread, it stops opening algorithms and starts the corresponding closing algorithm.
- Position updates and algorithm status callbacks keep the strategy’s displayed state current.
Tags
Full text
# BasicSpreadStrategy
# BasicSpreadStrategy
## Source (MIT)
```python
from datetime import datetime
from howtrader.app.spread_trading import (
SpreadStrategyTemplate,
SpreadAlgoTemplate,
SpreadData,
OrderData,
TradeData
)
class BasicSpreadStrategy(SpreadStrategyTemplate):
""""""
author = "用Python的交易员"
buy_price = 0.0
sell_price = 0.0
cover_price = 0.0
short_price = 0.0
max_pos = 0.0
payup = 10
interval = 5
start_time = "9:00:00"
end_time = "15:00:00"
spread_pos = 0.0
update_time = None
buy_algoid = ""
sell_algoid = ""
short_algoid = ""
cover_algoid = ""
parameters = [
"buy_price",
"sell_price",
"cover_price",
"short_price",
"max_pos",
"payup",
"interval"
]
variables = [
"spread_pos",
"update_time",
"buy_algoid",
"sell_algoid",
"short_algoid",
"cover_algoid",
]
def __init__(
self,
strategy_engine,
strategy_name: str,
spread: SpreadData,
setting: dict
):
""""""
super().__init__(
strategy_engine, strategy_name, spread, setting
)
self.start_t = datetime.strptime(self.start_time, "%H:%M:%S").time()
self.end_t = datetime.strptime(self.end_time, "%H:%M:%S").time()
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
def on_start(self):
"""
Callback when strategy is started.
"""
self.write_log("策略启动")
def on_stop(self):
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")
self.update_time = None
self.buy_algoid = ""
self.sell_algoid = ""
self.short_algoid = ""
self.cover_algoid = ""
self.put_event()
def on_spread_data(self):
"""
Callback when spread price is updated.
"""
# Trading is only allowed within given start/end time range
self.update_time = self.spread.datetime.time()
if self.update_time < self.start_t or self.update_time >= self.end_t:
self.stop_open_algos()
self.stop_close_algos()
self.put_event()
return
self.spread_pos = self.get_spread_pos()
# No position
if not self.spread_pos:
self.stop_close_algos()
# Start open algos
if not self.buy_algoid:
self.buy_algoid = self.start_long_algo(
self.buy_price, self.max_pos, self.payup, self.interval
)
if not self.short_algoid:
self.short_algoid = self.start_short_algo(
self.short_price, self.max_pos, self.payup, self.interval
)
# Long position
elif self.spread_pos > 0:
self.stop_open_algos()
# Start sell close algo
if not self.sell_algoid:
self.sell_algoid = self.start_short_algo(
self.sell_price, self.spread_pos, self.payup, self.interval
)
# Short position
elif self.spread_pos < 0:
self.stop_open_algos()
# Start cover close algo
if not self.cover_algoid:
self.cover_algoid = self.start_long_algo(
self.cover_price, abs(
self.spread_pos), self.payup, self.interval
)
self.put_event()
def on_spread_pos(self):
"""
Callback when spread position is updated.
"""
self.spread_pos = self.get_spread_pos()
self.put_event()
def on_spread_algo(self, algo: SpreadAlgoTemplate):
"""
Callback when algo status is updated.
"""
if not algo.is_active():
if self.buy_algoid == algo.algoid:
self.buy_algoid = ""
elif self.sell_algoid == algo.algoid:
self.sell_algoid = ""
elif self.short_algoid == algo.algoid:
self.short_algoid = ""
else:
self.cover_algoid = ""
self.put_event()
def on_order(self, order: OrderData):
"""
Callback when order status is updated.
"""
pass
def on_trade(self, trade: TradeData):
"""
Callback when new trade data is received.
"""
pass
def stop_open_algos(self):
""""""
if self.buy_algoid:
self.stop_algo(self.buy_algoid)
if self.short_algoid:
self.stop_algo(self.short_algoid)
def stop_close_algos(self):
""""""
if self.sell_algoid:
self.stop_algo(self.sell_algoid)
if self.cover_algoid:
self.stop_algo(self.cover_algoid)
```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.