一、策略介绍
1. 多周期趋势确认+短周期择时
利用了1小时周期均线差与 ATR 的比例判断大趋势方向（trend_direction）。
在趋势方向明确后，才允许在5分钟周期用“分型”做入场判断，有效过滤震荡噪声。
2. 使用分型形态作为入场条件
顶/底分型是经典的技术分析形态，结合趋势方向，有助于提高入场信号的胜率。
进一步要求分型后一根K线的收盘价确认，增强信号可靠性。
3. 每日只开一次仓，避免过度交易
使用 last_trade_date 限制每天只开一次仓，特别适用于日内稳健交易。
二、回测结果
爆仓了。
三、代码
from vnpy_ctastrategy import (
CtaTemplate,
BarData,
OrderData,
TradeData,
ArrayManager,
BarGenerator,
)
from vnpy.trader.constant import Interval, Direction, Offset, Status
from datetime import time, datetime
class SimpleFenxingStrategy(CtaTemplate):
author = "ChatGPT"
fast_ma_window = 10
slow_ma_window = 40
atr_length = 14
pricetick = 1
fixed_size = 1
parameters = [
"fast_ma_window",
"slow_ma_window",
"atr_length",
"pricetick",
"fixed_size",
]
variables = [
"trend_direction", # 1: 上升趋势，-1: 下降趋势，0: 无趋势
"last_trade_date", # 记录上次开仓日期，控制每天只开一次仓
]
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
self.bg_1h = BarGenerator(self.on_bar, window=12, on_window_bar=self.on_1h_bar, interval=Interval.MINUTE)
self.am_1h = ArrayManager(100)
self.bg_5m = BarGenerator(self.on_bar, window=5, on_window_bar=self.on_5m_bar, interval=Interval.MINUTE)
self.am_5m = ArrayManager(150)
self.bars_5m = [] # 缓存5分钟bar用于分型判断
self.trend_direction = 0
self.last_trade_date = None
def on_init(self):
self.load_bar(50)
def on_start(self):
self.write_log("策略启动")
def on_stop(self):
self.write_log("策略停止")
def on_bar(self, bar: BarData):
# 过滤非交易时间（根据实际合约调整）
bar_time = bar.datetime.time()
# 示例：过滤早盘9:00-9:30和下午13:30-14:45的时间段（如果需要）
if (time(9, 0) <= bar_time <= time(9, 30)) or (time(13, 30) <= bar_time <= time(14, 45)):
return
self.bg_1h.update_bar(bar)
self.bg_5m.update_bar(bar)
def on_1h_bar(self, bar: BarData):
self.am_1h.update_bar(bar)
if not self.am_1h.inited:
return
fast_ma = self.am_1h.sma(self.fast_ma_window)
slow_ma = self.am_1h.sma(self.slow_ma_window)
atr = self.am_1h.atr(self.atr_length)
if atr == 0:
self.trend_direction = 0
return
diff = fast_ma - slow_ma
if abs(diff) > 0.35 * atr:
self.trend_direction = 1 if diff > 0 else -1
else:
self.trend_direction = 0
def on_5m_bar(self, bar: BarData):
self.am_5m.update_bar(bar)
if not self.am_5m.inited:
return
self.bars_5m.append(bar)
if len(self.bars_5m) > 10:
self.bars_5m.pop(0)
if len(self.bars_5m) < 7:
return
if self.trend_direction == 0:
# 无趋势，不交易
return
today = bar.datetime.date()
# 每天最多开一次仓，且目前无仓位才允许开仓
if self.pos == 0 and self.last_trade_date == today:
return # 当天已开过仓，跳过
def is_top_fenxing(idx):
if idx < 2 or idx > len(self.bars_5m) - 3:
return False
return (
self.bars_5m[idx].high_price > self.bars_5m[idx - 1].high_price and
self.bars_5m[idx].high_price > self.bars_5m[idx - 2].high_price and
self.bars_5m[idx].high_price > self.bars_5m[idx + 1].high_price and
self.bars_5m[idx].high_price > self.bars_5m[idx + 2].high_price
)
def is_bottom_fenxing(idx):
if idx < 2 or idx > len(self.bars_5m) - 3:
return False
return (
self.bars_5m[idx].low_price < self.bars_5m[idx - 1].low_price and
self.bars_5m[idx].low_price < self.bars_5m[idx - 2].low_price and
self.bars_5m[idx].low_price < self.bars_5m[idx + 1].low_price and
self.bars_5m[idx].low_price < self.bars_5m[idx + 2].low_price
)
idx = len(self.bars_5m) - 4
if idx < 2:
return
# 多仓逻辑（趋势向上）
if self.trend_direction == 1:
# 有仓位，遇顶分型，平仓
if self.pos > 0 and is_top_fenxing(idx):
self.sell(bar.close_price - self.pricetick, abs(self.pos))
self.write_log(f"趋势向上，形成顶分型，平多仓，价格:{bar.close_price}")
return
# 无仓且满足底分型，且分型后一根K线收盘价大于分型收盘价，开多仓
if (
self.pos == 0
and is_bottom_fenxing(idx)
and self.bars_5m[idx + 1].close_price > self.bars_5m[idx].close_price
):
self.buy(bar.close_price + self.pricetick, self.fixed_size)
self.last_trade_date = today
self.write_log(f"趋势向上，底分型开多仓，价格:{bar.close_price}")
return
# 空仓逻辑（趋势向下）
elif self.trend_direction == -1:
# 有空仓位，遇底分型，平仓
if self.pos < 0 and is_bottom_fenxing(idx):
self.cover(bar.close_price + self.pricetick, abs(self.pos))
self.write_log(f"趋势向下，形成底分型，平空仓，价格:{bar.close_price}")
return
# 无仓且满足顶分型，且分型后一根K线收盘价小于分型收盘价，开空仓
if (
self.pos == 0
and is_top_fenxing(idx)
and self.bars_5m[idx + 1].close_price < self.bars_5m[idx].close_price
):
self.short(bar.close_price - self.pricetick, self.fixed_size)
self.last_trade_date = today
self.write_log(f"趋势向下，顶分型开空仓，价格:{bar.close_price}")
return
def on_order(self, order: OrderData):
if order.status == Status.ALLTRADED:
direction = "买入" if order.direction == Direction.LONG else "卖出"
self.write_log(f"{direction}订单成交 {order.volume}手 @{order.price}")
def on_trade(self, trade: TradeData):
if trade.offset == Offset.OPEN:
self.write_log(f"开仓成交 {trade.direction.value} {trade.volume}手 @{trade.price}")
elif trade.offset == Offset.CLOSE:
if self.pos == 0:
self.write_log("平仓完成，仓位清零")