这是我给DEEPSEEK的指令
我在用同花顺旗下的supermind进行交易策略回测。请根据我的策略，给我一份完整的可以在supermind进行回测的代码。 我的交易策略如下：股票池挑选为：前一日涨停的股票，且最近一个月内有过涨停历史，并且仅限于首板的股票。买入条件为：非一字板外，第一个涨停的股票。买入时间为：涨停封板的一瞬间排单买入。 卖出条件为：次日集合竞价结束在-3%以下，开盘不论任何价位立马卖出。持仓股票次日上午10点30分之前未涨停，卖出。 持仓股票次日上午10点之前不是红盘卖出。 如果10点半之前涨停，则继续持有。，并且实时监控股票动向，若当天涨停板打开，即刻卖出。 仓位控制：每次买入为50%仓位，一天只进行一只股票的操作。如此策略下滚动操作。当满仓后，股票都没有达到卖出条件则一直持有等待出现卖出信号
这是DEEPSEEK给我代码：
# -*- coding: utf-8 -*-
from supermind.api import *
def initialize(context):
# ========== 策略参数配置 ==========
g.trade_count = 0 # 当日交易计数
g.holding = {} # 持仓记录
g.limit_history = {} # 涨停历史记录
# 策略参数
set_params(
position_ratio = 0.5, # 仓位比例
check_interval = '1m', # 监控频率
limit_up_days = 30, # 涨停历史周期
max_stocks = 1 # 最大持仓数
)
# 定时任务设置
schedule(
time_open_pre, # 开盘前运行
time_rules.market_open(minutes=30), # 提前30分钟准备
)
schedule(
execute_trading, # 盘中交易执行
time_rules.every_minute(),
)
# ========== 核心策略函数 ==========
def time_open_pre(context):
"""开盘前准备"""
clear_signals()
g.trade_count = 0
g.today_limit_pool = get_limit_pool(context)
def get_limit_pool(context):
"""获取有效涨停股票池"""
# 获取前日涨停股
hist = history(
field='is_limit_up',
security=list(get_all_securities('stock').index),
count=1,
frequency='1d',
)
# 筛选首板非一字板
valid_stocks = []
for stock in hist.columns:
# 排除ST/*ST
if 'ST' in get_security_info(stock).display_name:
continue
# 获取30日涨停记录
limit_days = history(
field='is_limit_up',
security=stock,
count=g.limit_up_days,
frequency='1d',
).sum()
# 首次涨停判断
if (hist[stock][-1] == 1 and
limit_days == 1 and
not is_straight_limit(stock)):
# 记录首板日期
g.limit_history[stock] = context.current_dt.date()
valid_stocks.append(stock)
return valid_stocks
def is_straight_limit(stock):
"""判断是否一字板"""
today = history(
field=['open','high','low','close'],
security=stock,
count=1,
frequency='1d',
)
return (today['open'][-1] == today['high'][-1] and
today['close'][-1] == today['high'][-1])
def execute_trading(context):
"""交易执行主逻辑"""
current_time = context.current_dt.time()
# ===== 买入逻辑 =====
if g.trade_count == 0 and len(g.holding) < g.max_stocks:
for stock in g.today_limit_pool:
# 涨停瞬间检测
if check_limit_up_instant(stock):
# 计算仓位
cash = context.portfolio.cash
price = get_current_price(stock)
amount = int((cash * g.position_ratio) // (price * 100)) * 100
# 发送买入订单
order(stock, amount)
g.holding[stock] = {
'buy_time': context.current_dt,
'sold': False
}
g.trade_count += 1
break
# ===== 卖出逻辑 =====
for stock in list(g.holding.keys()):
if g.holding[stock]['sold']:
continue
# 持仓信息获取
position = context.portfolio.positions[stock]
buy_date = g.holding[stock]['buy_time'].date()
current_date = context.current_dt.date()
# 次日判断
if current_date > buy_date:
# 条件1：10点前非红盘
if current_time < time(10,0):
if get_current_price(stock) < position.avg_cost:
order_target(stock, 0)
g.holding[stock]['sold'] = True
# 条件2：10:30前未涨停
elif time(10,0) <= current_time < time(10,30):
if not check_limit_status(stock):
order_target(stock, 0)
g.holding[stock]['sold'] = True
# 条件3：开板检测
else:
if check_limit_opened(stock):
order_target(stock, 0)
g.holding[stock]['sold'] = True
# ========== 工具函数 ==========
def check_limit_up_instant(stock):
"""涨停瞬间检测（分笔数据验证）"""
tick = get_current_tick(stock)
return (tick.current > tick.limit_up * 0.997 and
(tick.current - tick.pre_close)/tick.pre_close > 0.095)
def check_limit_status(stock):
"""当前是否封板"""
return get_current_price(stock) >= get_limit_price(stock)
def check_limit_opened(stock):
"""涨停板是否打开"""
hist = history(
field='high',
security=stock,
count=5,
frequency='1m',
)
return hist[-1] < get_limit_price(stock)
def get_limit_price(stock):
"""计算涨停价"""
pre_close = history(
field='close',
security=stock,
count=1,
frequency='1d',
).iloc[-1]
return round(pre_close * 1.1, 2)
# ========== 可视化分析 ==========
def analyze(context, records):
# 策略绩效分析
plot_portfolio(records)
plot_benchmark_comparison(records)
# 涨停参与度分析
limit_participation = pd.Series({
'有效信号': len(g.today_limit_pool),
'实际成交': g.trade_count
})
limit_participation.plot.pie(title='涨停参与度')
