没有过拟合，没有未来函数，熬了无数个深夜，想破脑袋的策略，我觉得不完美，还得继续优化，找出规律，主要代码如下：
集合竞价，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"
