龙门大侠：https://quant.10jqka.com.cn/view/user.html#/user/714639531

正常：竞价涨幅>=2%且<=5.5%，竞价量/成交量>=3%，涨停封单量/成交量>=4%，个股热度排名<=150，换手率>=10%且<=30%，至的涨跌幅>=15%且<=60%

回测：2025-02-05至2025-04-30收益421.68%，2025-02-05至2025-03-27收益312.49%，2025-03-28至2025-04-30收益-13.06%

弱市：竞价涨幅>=0%且<=4%，竞价量/成交量>=3%，涨停封单量/成交量>=1%，个股热度排名<=100，换手率>=10%且<=25%，至的涨跌幅>=5%且<=40%

回测：2025-02-05至2025-04-30收益-3.93%，2025-02-05至2025-03-27收益-18.62%，2025-03-28至2025-04-30收益28.45%

大盘强弱切换：2025-02-05至2025-05-12收益646.03%，2025-02-05至2025-03-27收益312.49%，2025-03-28至2025-05-12收益40.53%

from datetime import datetime, timedelta
import pandas as pd
from tabulate import tabulate
from mindgo_api import *

def init(context):
context.holding_period = 7 # 股票持有天数（7天）
context.stop_gain = 0.8 # 止盈涨幅阈值（80%）
context.stop_gain_drawdown = 0.07 # 止盈回撤阈值（7%）
context.stop_loss = 0.07 # 止损跌幅阈值（7%）
context.position_pct = 1 # 仓位比例（100%）
context.information = {} # 存储股票额外信息
context.stock_data = {} # 存储股票数据
context.cached_history = {} # 历史数据缓存
set_commission(PerShare(type='stock', cost=0.00022, min_trade_cost=5)) # 设置交易佣金（万2.2）
set_slippage(PriceSlippage(0.005)) # 设置交易滑点（0.5%）
run_daily(close_sell, 'before_close', 0, 5, '000001.SZ') # 每天收盘前5分钟执行

每天集合竞价后(9:26)，调用一次open_auction函数，更新股票数据

def open_auction(context, bar_dict):
dates = get_dates() # 获取相关日期
market_trend = get_market_trend() # 判断市场趋势
query = get_stock_wencai(market_trend, dates) # 获取选股条件
try:
result = query_iwencai(query, domain='股票', df=True) # 执行选股查询
if result.empty:
print(f"未找到符合条件的股票") # 无符合条件股票
return
print(f"获取数据：{len(result)}条") # 打印结果数量
print(f"当前市场趋势：{'上升' if market_trend == 'up' else '下降'}") # 打印市场趋势
open_sell(context) # 执行开盘卖出
buy_stocks(context, result) # 执行买入操作
context.stock_data[dates[0]] = result # 存储当天股票数据
except Exception as e:
print(f"查询数据时出错：{e}") # 异常处理

问财查询函数

def get_stock_wencai(market_trend, dates):
today, yesterday, pastday2, pastday5, pastday10, pastday11, pastday21 = dates
if market_trend == "up":
query = (
f"{today}竞价涨幅>=2%且<=5.5%，" # 竞价涨幅范围
f"{today}竞价量/{yesterday}成交量>=3%，" # 竞价量占比
f"{yesterday}成交量>{pastday11}至{pastday2}的日均成交量，" # 成交量放大条件
f"{yesterday}成交量>{pastday21}至{pastday2}的最高量0.5，" # 成交量突破条件
f"{yesterday}涨停封单量/{yesterday}成交量>=4%，" # 封单量要求
f"{yesterday}个股热度排名<=150，" # 热度排名限制
f"{yesterday}换手率>=10%且<=30%，" # 换手率范围
f"{yesterday}的概念龙头，" # 必须是概念龙头
f"{pastday10}至{yesterday}的涨跌幅>=15%且<=60%，" # 近期涨幅范围
"非st，非创业板，非科创板，非北证a股" # 排除特定板块
)
else:
query = (
f"{today}竞价涨幅>=0%且<=4%，" # 弱市竞价涨幅范围
f"{today}竞价量/{yesterday}成交量>=3%，" # 竞价量占比
f"{yesterday}成交量>{pastday11}至{pastday2}的日均成交量，" # 成交量放大条件
f"{yesterday}成交量>{pastday21}至{pastday2}的最高量0.5，" # 成交量突破条件
f"{yesterday}涨停封单量/{yesterday}成交量>=1%，" # 封单量要求
f"{yesterday}个股热度排名<=100，" # 热度排名限制
f"{yesterday}换手率>=10%且<=25%，" # 换手率范围
f"{yesterday}的概念龙头，" # 必须是概念龙头
f"{pastday10}至{yesterday}的涨跌幅>=5%且<=40%，" # 近期涨幅范围
"非st，非创业板，非科创板，非北证a股" # 排除特定板块
)
return query

买入股票的函数

def buy_stocks(context, stocks_list):
if not stocks_list.empty:
cash_per_stock = context.portfolio.cash / len(stocks_list) # 计算每只股票分配资金
today = get_dates()[0] # 获取当前日期
for stock in stocks_list['股票代码']:
try:
order_value(stock, cash_per_stock) # 按价值下单
context.information[stock] = Information(context.holding_period) # 记录股票信息
context.information[stock].buy_date = today # 记录买入日期
print(f"买入股票：{stock}，买入金额：{cash_per_stock:.2f}") # 打印买入信息
except Exception as e:
print(f"买入股票 {stock} 失败：{e}") # 买入失败处理

开盘时卖出股票（昨日未涨停的股票）

def open_sell(context):
for stock in list(context.portfolio.positions):
try:
y_data = history(stock, ['close', 'high_limit'], 1, '1d', True, 'pre').iloc[0] # 获取昨日行情数据
y_close = y_data['close']
y_limit_up = y_data['high_limit'] # 昨日涨停价
if abs(y_close - y_limit_up) < y_limit_up * 0.005: # 判断昨日是否涨停（收盘价是否接近涨停价）
print(f"开盘保留 {stock}（昨日涨停）| 涨停价: {y_limit_up:.2f} | 昨收: {y_close:.2f}")
else:
order_target(stock, 0)
print(f"开盘卖出 {stock}（昨日未涨停）| 涨停价: {y_limit_up:.2f} | 昨收: {y_close:.2f}")
except Exception as e:
print(f"开盘卖出 {stock} 出错：{e}")

收盘时卖出股票（收盘价<最高价，当日未保持强势）

def close_sell(context, bar_dict):
current_time = get_datetime()
for stock in list(context.portfolio.positions):
try:
ticks = get_tick(stock, current_time.replace(hour=9, minute=30), current_time, ['current'])
current_price = ticks.iloc[-1]['current'] # 当前价
current_high = bar_dict[stock].high # 今日最高价
current_high_limit = bar_dict[stock].high_limit # 今日涨停价
if current_price <= current_high * 0.995: # 当前价<最高价，当日未保持强势，则卖出
order_target(stock, 0)
print(f"收盘卖出 {stock}（当日未保持强势）| 涨停价:{current_high_limit:.2f} 今高:{current_high:.2f} 现价:{current_price:.2f}")
except Exception as e:
print(f"收盘卖出 {stock} 出错：{e}")

判断大盘趋势

def get_market_trend():
sz_index = history('399001.SZ', ['close'], 22, '1d', True, 'pre') # 获取深证成指的数据
if len(sz_index) < 20:
return "neutral" # 数据不足返回中性
sz_ma10 = sz_index['close'].rolling(10).mean().iloc[-1] # 计算10日均线
sz_ma20 = sz_index['close'].rolling(20).mean().iloc[-1] # 计算20日均线
if sz_ma10 > sz_ma20:
return "up"
else:
return "down"

排除非交易日

def get_dates():
today = get_datetime().strftime('%Y-%m-%d') # 当前日期
dates = [today] # 初始化日期列表
for days in [1, 2, 5, 10, 11, 21]: # 需要获取的历史交易日
past_date = get_datetime()
for _ in range(days):
past_date = get_previous_trading_date(past_date) # 获取前一个交易日
dates.append(past_date.strftime('%Y-%m-%d')) # 添加到日期列表
return tuple(dates) # 返回日期元组

记录买入股票的信息，作为position的补充

class Information:
def init(self, remain_days):
self.remain_days = remain_days # 剩余持有天数
self.stop_gain = False # 是否进入止盈状态
self.today_high = 0 # 今日最高价
self.high = 0 # 止盈时的最高价