研究了两年，终于研究出来一个无敌策略，不惧牛熊，各种行情都是稳定盈利！！ 有感兴趣的朋友欢迎留言，短周期策略。持仓数量十只 之前我分享过一个小工具网站，支持国内主流量化平台，可以让 AI 直接帮你写各个平台的策略代码，直接生成可运行的策略代码，代码质量远高于直接使用 DeepSeek、Trae 等平台。上线之后获得了非常多朋友的好评。 大家可以直接用描述策略，然后一键生成可运行的完整策略代码，也可以把它当做一个API 查询工具。 AI工具平台：https://iris.findtruman.io/ai/tool/ai-quantitative-trading/ 我看平台正在开发SuperMind支持，很快就能支持同花顺了 回测日期：2014-12-5 至 2025-05-31 选股策略： 选取中证1000指数中的1000只股票 当日涨跌幅 3-5 当日换手率 3-10 当日均线 MA5 > MA10 止损止盈条件 收益 > 5 或者 收益-3卖出 # 中证1000选股策略 # 作者 但盼风雨来 # 选股策略： # 当日涨跌幅 3-5 # 当日换手率 3-10 # 当日均线 MA5 > MA10 # 止损止盈条件 收益 > 5 或者 收益-3卖出 import pandas as pd # 初始化函数,全局只运行一次 def init(context): # 设置基准收益：沪深300指数 set_benchmark('000300.SH') # 打印日志 log.info('策略开始运行,初始化函数全局只运行一次') # 设置股票每笔交易的手续费为万分之二(手续费在买卖成交后扣除,不包括税费,税费在卖出成交后扣除) set_commission(PerShare(type='stock',cost=0.0000854,min_trade_cost=0)) # 设置股票交易双边滑点0.5%,表示买入价为实际价格乘1.0025,卖出价为实际价格乘0.9975 set_slippage(PriceSlippage(0.005)) # 设置日级最大成交比例25%,分钟级最大成交比例50% # 日频运行时，下单数量超过当天真实成交量25%,则全部不成交 # 分钟频运行时，下单数量超过当前分钟真实成交量50%,则全部不成交 set_volume_limit(0.25,0.5) # 设置要操作的股票：同花顺 context.security = ['300033.SZ'] # 回测区间、初始资金、运行频率请在右上方设置 context.stock_list = ['300033.SZ','000718.SZ'] run_daily(function) #每日开盘前9:00被调用一次,用于储存自定义参数、全局变量,执行盘前选股等 def before_trading(context): # 获取日期 date = get_datetime().strftime('%Y-%m-%d %H:%M:%S') # 打印日期 log.info('{} 盘前运行'.format(date)) ## 开盘时运行函数 def handle_bar(context, bar_dict): # 获取时间 time = get_datetime().strftime('%Y-%m-%d %H:%M:%S') log.info(bar_dict) # 买卖策略 positions = context.stock_account.positions to_sell = [] for position in positions: shouyi = positions[position].profit_rate * 100 # log.info(f"{positions[position].symbol}的收益率为{shouyi}%") stock = positions[position].symbol if shouyi > 6 or shouyi <-3: to_sell.append(stock) for stock in to_sell: order_target(stock,0) log.warn(f"已清仓了{stock}") # 打印时间 log.info('{} 盘中运行'.format(time)) stock_list = context.stock_list log.info("下单的股票为",str(stock_list)) for stock in stock_list: price = get_current(stock) current_price = price[stock].open # log.info(f"股票代码{stock},价格{current_price}") order(stock,1000,price = current_price) date = get_datetime() # log.info('{} 运行'.format(date)) log.info(f"已购买{stock},价格为{current_price}") ## 收盘后运行函数,用于储存自定义参数、全局变量,执行盘后选股等 def after_trading(context): # 获取时间 time = get_datetime().strftime('%Y-%m-%d %H:%M:%S') # 打印时间 log.info('{} 盘后运行'.format(time)) log.info('一天结束') def function(context,*args): get_all_stock(context) get_rate_stock(context) change_hand(context) avg_stock(context) # 获取所有股票 def get_all_stock(context): # 获取时间 time = get_datetime() # 获取市场所有股票 # all_stock = get_all_securities(ty='stock', date=time) # df = all_stock # df = df[~df['display_name'].str.contains('ST', na=False)] # df = df[~df['order_book_id'].str.contains('BJ')] # stock_list = df['order_book_id'].tolist() # 获取中证1000 stock_list= get_index_stocks('000852.SH',time) log.info("已获取到所有股票代码") context.stock_list = stock_list log.info(f"中证1000股票数量为{len(stock_list)}") log.info("已获取中证1000股票") # 筛选符合涨幅条件的股票 def get_rate_stock(context): time = get_datetime().strftime('%Y-%m-%d') stock_list = context.stock_list null_list = [] fields = ['quote_rate','open'] current = get_price( stock_list, start_date=None, end_date=time, fre_step='1d', fields=fields, skip_paused=False, fq='pre', bar_count=1, is_panel=False, ) for stock in stock_list: if (current[stock].quote_rate[0] > 3) & (current[stock].quote_rate[0] < 5): null_list.append(stock) context.stock_list = null_list log.info(f"符合涨跌幅限制的股票数量为{len(null_list)}") log.info("执行筛选符合涨幅条件的股票") # 筛选符合换手率的股票 def change_hand(context): current_stock = context.stock_list hand_list = [] time = get_datetime().strftime('%Y%m%d') fields = ['turnover_rate'] change_hand = get_price( current_stock, start_date=None, end_date=time, fre_step='1d', fields=fields, skip_paused=False, fq='pre', bar_count=1, is_panel=False, ) for stock in current_stock: if (float(change_hand[stock]['turnover_rate'])>3)&(float(change_hand[stock]['turnover_rate'])<10): hand_list.append(stock) context.stock_list = hand_list log.info(f"符合换手率的股票数量为{len(hand_list)}") log.info("执行筛选符合换手率的股票") # 筛选符合均线的股票 def avg_stock(context): time = get_datetime().strftime('%Y%m%d') stock_list = context.stock_list avg_list = [] price = get_price( stock_list, start_date=None, end_date=time, fre_step='1d', fields=['close'], skip_paused=False, fq='pre', bar_count=20, is_panel=False, ) for stock in stock_list: MA5 = price[stock]['close'].iloc[-5:].mean() MA10 = price[stock]['close'].iloc[-10:].mean() if MA5 > MA10 : avg_list.append(stock) context.stock_list = avg_list log.info(f"符合均线策略的股票数量为{len(avg_list)}") log.info("执行筛选符合均线的股票") 龙门大侠：https://quant.10jqka.com.cn/view/user.html#/user/714639531 大盘切换回测效果：2025-02-05至2025-05-12收益646.03%，2025-02-05至2025-03-27收益312.49%，2025-03-28至2025-05-12收益40.53%。 欢迎抛砖引玉，指点迷津，问财查询函数query_iwencai，不用get_iwencai是因为我当天参考他的选股，通过custom_sector更新股票到同花顺软件的自定义板块。 开盘卖出昨天未涨停的股票，腾出资金，根据大盘强弱切换选股条件，开盘买入符合条件的股票。 后续的版本添加盘中检查移动止损并预测连板可能性，预测次日是否能够连板，识别主力出货，判断最佳卖出时机。 策略收益 基准收益 策略年化收益率 基准年化收益 Alpha Beta Sharpe Sortino Information Ratio Volatility 最大回撤 Tracking Error Downside Risk 胜率 次胜率 646.03% 1.93% 227285.35% 7.61% 2272.76 1.31 2193.42 4323.95 7.41 1.04 20.26% 1 0.53 60.27% 76.19% 强弱切换代码： result = query_iwencai(query, domain='股票', df=True) # 执行选股查询 query = ( f"{today}竞价涨幅>=2%且<=5.5%，" # 竞价涨幅范围 f"{today}竞价量/{yesterday}成交量>=3%，" # 竞价量占比 "非st，非创业板，非科创板，非北证a股" # 排除特定板块 ) else: query = ( f"{today}竞价涨幅>=0%且<=4%，" # 弱市竞价涨幅范围 f"{today}竞价量/{yesterday}成交量>=3%，" # 竞价量占比 ) return query 以下是我正在实现的功能： ====== ====== ====== 识别主力出货 ====== ====== ====== 拉高出货特征（完全重构计算逻辑） 1.1 快速拉升后大量成交 1.2 高位频繁震荡（使用numpy提高性能） 资金流向特征（完全重写） 2.1 大单卖出比例 2.2 主力资金持续流出 盘口特征（优化订单簿分析） 3.1 虚假买盘检测 决策输出（添加时间过滤） ====== ====== ====== 判断最佳卖出时机 ====== ====== ====== 时间段评分 价格走势评分 成交量评分 大单分析 ====== ====== ====== 检查移动止损并预测连板可能性 ====== ====== ====== 每5分钟评估一次连板强度 检查主力出货 常规移动止损检查 
没有过拟合，没有未来函数，熬了无数个深夜，想破脑袋的策略，我觉得不完美，还得继续优化，找出规律，主要代码如下： 集合竞价，9:26自动运行 def open_auction(context, bar_dict): try: get_open_sell(context, bar_dict) # 开盘卖出低开-9%的股票 get_auction_analysis(context) # 更新竞价强度 get_market_trend(context) # 缓存大盘趋势 print(f"大盘趋势: {'上升 🔴' if context.market_stop == 'up' else '下降 🟢'}") print(f"高开比例: {'强势 🟥' if context.limit_up_stop == 'up' else '弱势 🟩'}") if context.market_stop == 'down' and context.limit_up_stop == 'down': print("❌ 盘前检测到风险，今日禁止买入") return result = query_iwencai(query=get_stock_wencai(context.market_stop), domain='股票', df=True) if not result.empty: get_buy_stocks(context, result) except Exception as e: print(f"集合竞价 出错: {str(e)[:30]}") 构建问财查询语句 def get_stock_wencai(m_stop): dates = get_custom_dates(custom_days=[1, 2, 6]) today, yesterday, pastday2, pastday6 = dates if m_stop == "up": query = ( f"{today}竞价涨幅>=2%且<10%，" f"{today}竞价未匹配金额，" f"{today}竞价量/{yesterday}成交量>=2%，" f"{yesterday}成交量>{pastday6}至{pastday2}的日均成交量，" f"{yesterday}个股热度排名<=200，" f"{yesterday}的20日均线角度>20，" f"{yesterday}的10日均线>20日均线，" "非st，非创业板，非科创板，非北证a股" ) else: query = ( f"{today}竞价涨幅>=0%且<10%，" f"{today}竞价未匹配金额，" f"{today}竞价量/{yesterday}成交量>=1%，" f"{yesterday}成交量>{pastday6}至{pastday2}的日均成交量，" f"{yesterday}个股热度排名<=150，" f"{yesterday}的20日均线角度>20，" f"{yesterday}的10日均线>20日均线，" "非st，非创业板，非科创板，非北证a股" ) return query 获取热门概念成分股 def get_hot_concept(context, date): try: result_zhishu = query_iwencai(f'{date}同花顺概念指数涨幅，{date}涨停家数占比', domain='指数', df=True) excluded_concepts = ['融资融券', '深股通', '沪股通', '国企改革', '央企国企改革', '专精特新'] result_zhishu = result_zhishu[~result_zhishu['指数简称'].isin(excluded_concepts)] n = 10 if context.market_stop == "up" else 3 # 根据市场趋势确定筛选数量 concept_range = result_zhishu.nlargest(n, '指数@涨跌幅:前复权') # 获取涨幅和上涨家数占比靠前的概念 concept_ratio = result_zhishu.nlargest(n, '指数@涨停家数占比') if context.market_stop == "up": concept_all = pd.merge(concept_range, concept_ratio, how='inner', on=['指数代码', '指数简称']) # 强势市场取交集 else: concept_all = pd.concat([concept_range, concept_ratio]).drop_duplicates(subset=['指数代码', '指数简称']) # 弱势市场取并集 concept_stocks = {} for concept_code in concept_all['指数代码'].unique(): stocks = get_concept_stocks(concept_code, date=date) # 获取概念成分股 if stocks: concept_stocks[concept_code] = stocks return concept_stocks except Exception as e: print(f"获取热门概念成分股 出错: {str(e)[:30]}") return {} 集合竞价分析 def get_auction_analysis(context): t_date = get_datetime().strftime('%Y%m%d') # 当前交易日 y_date = get_previous_trading_date(get_datetime()).strftime('%Y%m%d') # 前一交易日 try: result = query_iwencai(query=f'{y_date}涨停，{t_date}竞价涨幅', domain='股票', df=True) if result.empty: context.limit_up_stop = "down" return limit_up_count = len(result) up_count = len(result[result['竞价涨幅'] > 0]) ratio = up_count / limit_up_count if limit_up_count > 0 else 0 if ratio >= 0.6: context.limit_up_stop = "up" else: context.limit_up_stop = "down" except Exception as e: print(f"集合竞价分析 出错: {str(e)[:30]}") context.limit_up_stop = "down" 判断大盘趋势 def get_market_trend(context): try: hs_index = history('399905.SZ', ['close', 'volume'], 20, '1d', True, 'pre') # 获取历史数据 if len(hs_index) < 20: context.market_stop = "normal" return hs_close = hs_index['close'] # 提取收盘价 hs_volume = hs_index['volume'] # 提取成交量 hs_ma5 = hs_close.rolling(5).mean().iloc[-1] # 计算均线 hs_ma10 = hs_close.rolling(10).mean().iloc[-1] hs_ma20 = hs_close.rolling(20).mean().iloc[-1] avg_volume_20 = hs_volume.rolling(20).mean().iloc[-1] # 计算成交量指标 last_volume = hs_volume.iloc[-1] close_prices = hs_close.values macd, signal, _ = talib.MACD(close_prices, fastperiod=6, slowperiod=12, signalperiod=5) # 计算MACD指标 is_macd_dead = macd[-1] < signal[-1] # DIFF < DEA 视为弱势 is_below_ma5 = hs_close.iloc[-1] < hs_ma5 # 是否跌破5日均线 if is_macd_dead and is_below_ma5: # 判断趋势 trend = "down" elif hs_ma10 > hs_ma20: if hs_ma5 < hs_ma10 and last_volume < avg_volume_20: trend = "down" else: trend = "up" else: trend = "down" context.market_stop = trend except Exception as e: print(f"判断大盘趋势 出错: {str(e)[:30]}") context.market_stop = "normal" 在模拟回测环境里 get_iwencai查询得到的股票 查看交易记录被推迟晚送了1天 比如应该9月1日选出的个股 9月3日才出现在iwencai_securities里 请问怎么解决这个问题 写一个小策略测试以下KD金叉的效率，新手上路，多多指教！！！ 
