用AI写代码，运行没问题，就是没有成交记录，后来排查才发现，代码的逻辑有问题。
真坑。
一、介绍
趋势过滤：使用 fast_ma 和 slow_ma 均线判断趋势方向。
突破入场：价格突破过去 entry_window 天最高/最低价后入场。
波动性过滤：趋势强度由 ma_diff > atr * coeff 判断，过滤震荡市。
ATR止损：以波动性设定止损，防止小波动被扫出。
只做多/空可选，适合不同品种。
二、回测结果
2025-05-23 19:45:57.106722 首个交易日： 2015-01-05
2025-05-23 19:45:57.107514 最后交易日： 2024-12-31
2025-05-23 19:45:57.110573 总交易日： 2488
2025-05-23 19:45:57.111033 盈利交易日： 689
2025-05-23 19:45:57.112491 亏损交易日： 704
2025-05-23 19:45:57.115781 起始资金： 20,000.00
2025-05-23 19:45:57.116271 结束资金： 38,361.71
2025-05-23 19:45:57.116766 总收益率： 91.81%
2025-05-23 19:45:57.117300 年化收益： 8.86%
2025-05-23 19:45:57.117805 最大回撤: -17,547.63
2025-05-23 19:45:57.118305 百分比最大回撤: -35.05%
2025-05-23 19:45:57.118767 最大回撤天数: 177
2025-05-23 19:45:57.119988 总盈亏： 18,361.71
2025-05-23 19:45:57.121777 总手续费： 1,408.29
2025-05-23 19:45:57.123712 总滑点： 3,820.00
2025-05-23 19:45:57.124429 总成交金额： 14,082,870.00
2025-05-23 19:45:57.124796 总成交笔数： 375
2025-05-23 19:45:57.125166 日均盈亏： 7.38
2025-05-23 19:45:57.125532 日均手续费： 0.57
2025-05-23 19:45:57.125938 日均滑点： 1.54
2025-05-23 19:45:57.126321 日均成交金额： 5,660.32
2025-05-23 19:45:57.128379 日均成交笔数： 0.15072347266881028
2025-05-23 19:45:57.129491 日均收益率： 0.03%
2025-05-23 19:45:57.129915 收益标准差： 1.30%
2025-05-23 19:45:57.130315 Sharpe Ratio： 0.31
2025-05-23 19:45:57.130708 EWM Sharpe： -0.28
2025-05-23 19:45:57.131101 收益回撤比： 2.62
2025-05-23 19:45:57.132166 策略统计指标计算完成
三、代码
import logging
import os
from datetime import time
import numpy as np
# 日志目录
log_dir = "./log"
os.makedirs(log_dir, exist_ok=True)
log_path = os.path.join(log_dir, "aberration_01.log")
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s: %(message)s",
handlers=[logging.FileHandler(log_path, mode='a', encoding='utf-8')],
)
from vnpy_ctastrategy import (
CtaTemplate, BarData, TickData, TradeData, BarGenerator
)
from vnpy.trader.utility import ArrayManager
from vnpy.trader.constant import Interval
class aberration_01(CtaTemplate):
author = "移植自你提供的策略"
entry_window = 10
exit_ma_window = 8
fast_ma = 10
slow_ma = 48
atr_window = 14
atr_multiplier = 1.0
trend_threshold = 0.05
fixed_size = 1
parameters = [
"entry_window", "exit_ma_window",
"fast_ma", "slow_ma",
"atr_window", "atr_multiplier",
"trend_threshold", "fixed_size"
]
variables = ["long_entry", "short_entry", "long_stop", "short_stop"]
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
self.am = ArrayManager(size=100)
self.bg = BarGenerator(self.on_bar, 1, self.on_daily_bar, Interval.DAILY, daily_end=time(hour=14, minute=59))
self.logger = logging.getLogger(strategy_name)
self.long_entry = 0.0
self.short_entry = 0.0
self.long_stop = 0.0
self.short_stop = 0.0
def on_init(self):
self.logger.info("策略初始化")
self.load_bar(100)
def on_start(self):
self.logger.info("策略启动")
def on_stop(self):
self.logger.info("策略停止")
def on_bar(self, bar: BarData):
self.bg.update_bar(bar)
def on_daily_bar(self, bar: BarData):
#确认是否被调用
self.logger.info(f"[on_hour_bar] 收到日K线: {bar.datetime}, 开盘价: {bar.open_price},收盘价: {bar.close_price},最高价: {bar.high_price},最低价: {bar.low_price}")
self.am.update_bar(bar)
if not self.am.inited:
return
high = self.am.high_array[-self.entry_window:-1]
low = self.am.low_array[-self.entry_window:-1]
close = self.am.close_array
atr = self.am.atr(self.atr_window)
fast = self.am.sma(self.fast_ma)
slow = self.am.sma(self.slow_ma)
trend_strength = abs(fast - slow)
# 趋势过滤
if trend_strength < self.trend_threshold * atr:
return
self.long_entry = np.max(high)
self.short_entry = np.min(low)
self.logger.info(f"当前收盘价: {bar.close_price}, 多单入场阈值: {self.long_entry}, 空单入场阈值: {self.short_entry}")
if self.pos == 0:
if bar.close_price > self.long_entry:
self.buy(bar.close_price + 1, self.fixed_size)
self.long_stop = bar.close_price - self.atr_multiplier * atr
self.logger.info(f"[开多] {bar.datetime} 价格: {bar.close_price}, 止损: {self.long_stop}")
elif bar.close_price < self.short_entry:
self.short(bar.close_price - 1, self.fixed_size)
self.short_stop = bar.close_price + self.atr_multiplier * atr
self.logger.info(f"[开空] {bar.datetime} 价格: {bar.close_price}, 止损: {self.short_stop}")
elif self.pos > 0:
if bar.close_price < self.long_stop or bar.close_price < self.am.sma(self.exit_ma_window):
self.sell(bar.close_price - 1, abs(self.pos))
self.logger.info(f"[平多] {bar.datetime} 价格: {bar.close_price}")
elif self.pos < 0:
if bar.close_price > self.short_stop or bar.close_price > self.am.sma(self.exit_ma_window):
self.cover(bar.close_price + 1, abs(self.pos))
self.logger.info(f"[平空] {bar.datetime} 价格: {bar.close_price}")
def on_tick(self, tick: TickData):
pass
def on_trade(self, trade: TradeData):
#self.logger.info(f"成交信息: {trade}")
pass
def on_order(self, order):
pass
def on_stop_order(self, order):
pass