def daily_selection(ContextInfo):
"""每日选股任务 - 根据新策略修改"""
清空候选股池
clear_sector(CANDIDATE_POOL_NAME)
candidate_stocks.clear()
# 从配置文件指定的通达信板块路径获取股票列表
stock_list = []
try:
fixed_path = ContextInfo.tdx_sector_path.replace('\\', '/')
with open(fixed_path, 'r', encoding='gbk') as f:
raw_list = [line.strip() for line in f.readlines() if line.strip()]
# 转换股票代码格式
for raw_code in raw_list:
try:
formatted_code = get_code(raw_code)
if formatted_code != raw_code:
stock_list.append(formatted_code)
else:
logger.log(f"忽略无效代码格式: {raw_code}", level="WARNING")
except Exception as e:
logger.log(f"代码转换失败: {raw_code} - {str(e)}", level="WARNING")
except Exception as e:
logger.log(f"读取通达信板块失败: {str(e)}", level="INFO")
return
for stock_code in stock_list:
try:
# 获取日线数据(获取15天数据确保有足够历史数据)
df_daily = ContextInfo.get_market_data_ex(
fields=["close", "high", "low", "volume", "open"],
stock_code=[stock_code],
count=15,
period="1d",
dividend_type="front"
).get(stock_code)
if df_daily is None or len(df_daily) < 12: # 至少需要15天数据
logger.log(f"股票{stock_code}数据不足,跳过", level="WARNING")
continue
# 计算5日均线
close_prices = np.array(df_daily["close"])
ma5 = mt.MA(close_prices, 5)
# === 条件1: 倍量阳线+5日均线拐头 ===
last_index = len(df_daily) - 1 # 最新K线索引(昨日)
prev_index = last_index - 1 # 前一交易日
# 检查倍量阳线
prev_vol = df_daily["volume"].iloc[prev_index]
curr_vol = df_daily["volume"].iloc[last_index]
is_positive = df_daily["close"].iloc[last_index] > df_daily["open"].iloc[last_index]
volume_ok = curr_vol > ContextInfo.volume_multiple * prev_vol
# 检查5日均线拐头
ma5_turn_up = ma5[last_index] > ma5[prev_index] and ma5[prev_index] < ma5[prev_index-1]
# 排除天量/巨量
not_extreme = not is_extreme_volume(df_daily, last_index)
# 如果不满足条件1,跳过
if not (volume_ok and is_positive and ma5_turn_up and not_extreme):
continue
# 记录倍量阳线最低价
mark_low = df_daily["low"].iloc[last_index]
# === 条件2: 10日内最低价K线及区间检查 ===
# 在倍量阳线前10个交易日内寻找最低收盘价K线
search_start = max(0, last_index - 10) # 确保不越界
search_end = last_index # 搜索到倍量阳线的前一天
# 找到最低收盘价K线
min_close = float('inf')
min_index = -1
for i in range(search_start, search_end):
if df_daily["close"].iloc[i] < min_close:
min_close = df_daily["close"].iloc[i]
min_index = i
# 如果没有找到符合条件的K线,跳过
if min_index == -1:
continue
# 记录最低价K线的最高价
mark_high = df_daily["high"].iloc[min_index]
# 检查区间内收盘价是否都低于mark_high
# 区间:从最低价K线(min_index)到倍量阳线的前一天(last_index-1)
condition2 = True
# 遍历最近9天
for offset in range(1, 10):
# 如果最低价K线位置在offset天之前,则检查该天收盘价是否小于标记最高价
if min_index <= (last_index - offset):
if df_daily["close"].iloc[last_index - offset] >= mark_high:
condition2 = False
break
# 如果不满足条件2,跳过
if not condition2:
continue
# === 新增条件: RSI指标和总股本条件 ===
# 获取周线RSI
week_rsi_ok = False
try:
df_week = ContextInfo.get_market_data_ex(
fields=["close"],
stock_code=[stock_code],
count=4, # 获取7周数据
period="1w",
dividend_type="front"
).get(stock_code)
if df_week is not None and len(df_week) >= 6:
week_close = np.array(df_week["close"])
week_rsi = mt.RSI(week_close, 6)
week_rsi_ok = week_rsi[-1] > 50 if len(week_rsi) > 0 else False
except:
week_rsi_ok = False
# 获取月线RSI
month_rsi_ok = False
try:
df_month = ContextInfo.get_market_data_ex(
fields=["close"],
stock_code=[stock_code],
count=2, # 获取2个月数据
period="1month",
dividend_type="front"
).get(stock_code)
if df_month is not None and len(df_month) >= 6:
month_close = np.array(df_month["close"])
month_rsi = mt.RSI(month_close, 6)
month_rsi_ok = month_rsi[-1] > 50 if len(month_rsi) > 0 else False
except:
month_rsi_ok = False
# 总股本条件(小于10亿股)
total_shares = get_instrument_detail(stock_code, "total_shares")
total_ok = total_shares is not None and total_shares < 1000000000 # 10亿股
# 检查RSI条件
rsi_ok = week_rsi_ok or month_rsi_ok
# 如果所有条件满足
if rsi_ok and total_ok :
# 记录倍量阳线最低价
mark_low = df_daily["low"].iloc[last_index]
# 加入候选股池
candidate_stocks[stock_code] = {
"mark_low": mark_low, # 倍量阳线最低价
"mark_high": mark_high, # 最低价K线的最高价
"select_date": datetime.today().strftime("%Y%m%d")
}
add_stock_to_sector_custom(CANDIDATE_POOL_NAME, stock_code)
logger.log(f"加入候选股: {stock_code}, 最低价K线最高价:{mark_high:.2f}, 倍量阳线最低价:{mark_low:.2f}")
except Exception as e:
logger.log(f"处理股票{stock_code}时出错: {str(e)}", level="INFO")
if len(candidate_stocks) > 0:
logger.log(f"选股完成,新增{len(candidate_stocks)}只候选股", level="INFO")
for stock_code, info in candidate_stocks.items():
logger.log(f"入选股票: {stock_code}, 倍量阳线最低价: {info['mark_low']:.2f}, 突破阈值: {info['mark_high']:.2f}", level="INFO")
else:
logger.log("选股完成,候选股池为空", level="INFO") # 修改后