本帖最后由 jiatengxian 于 2026-1-15 13:39 编辑
注：下述仅是个人观点，仅供参考
这是一个根据自己对股票的想法用（AI）帮忙分析出来的程序，
分析想法：
1、2025今年一进二成功率最大的存在哪些共性
2、帮我生成对应的python脚本
由于自己电脑性能问题，运行脚本需要时间有些长（3-10分钟）
因：包含akshare依赖，所以.exe文件比较大（akshare = 用 Python 免费获取各类金融市场公开数据的工具库。）
由于脚本较长就不在文档中直接提现了，下方附有python脚本生成的.exe 可以直接使用，
以下源码仅供参考
[Python] 纯文本查看 复制代码 import akshare as ak
import pandas as pd
import datetime
# 新函数：动态获取热门板块（概念或行业）
def get_hot_sectors(date, top_n=10, sector_type="概念"): # sector_type: "概念" or "行业"
indicator = "今日" # 可换 "5日"等
sector_param = f"{sector_type}资金流"
try:
df = ak.stock_sector_fund_flow_rank(indicator=indicator, sector_type=sector_param)
# 按主力净流入降序，取top_n名称
hot_sectors = df.sort_values(by='主力净流入', ascending=False).head(top_n)['板块名称'].tolist()
print(f"动态热门{sector_type} ({date}): {hot_sectors}")
return hot_sectors
except Exception as e:
print(f"获取热门{sector_type}失败: {e}，使用默认。")
return ['半导体', '海南', '机器人'] # fallback
# 定义函数获取历史涨停数据并计算一进二概率（历史保持固定combo）
def analyze_historical_prob(date_start='20240101', date_end=None):
if date_end is None:
date_end = datetime.datetime.now().strftime('%Y%m%d')
# 获取历史所有涨停池
dates = pd.date_range(start=date_start, end=date_end).strftime('%Y%m%d')
zt_df_list = []
for d in dates:
try:
df = ak.stock_zt_pool_em(date=d)
if not df.empty:
df['date'] = pd.to_datetime(d)
zt_df_list.append(df)
except:
pass
if not zt_df_list:
return None # 无历史数据，返回None
historical_zt = pd.concat(zt_df_list, ignore_index=True)
# 添加circ_mv (流通市值，单位亿)
historical_zt['circ_mv'] = historical_zt['流通市值'] / 100000000
# 排序并识别首板和次日zt
historical_zt = historical_zt.sort_values(['代码', 'date'])
historical_zt['prev_date'] = historical_zt.groupby('代码')['date'].shift(1)
historical_zt['is_first_board'] = historical_zt['prev_date'].isna() | (historical_zt['date'] - historical_zt['prev_date'] > pd.Timedelta(days=1))
historical_zt['next_date'] = historical_zt.groupby('代码')['date'].shift(-1)
historical_zt['next_day_zt'] = (historical_zt['next_date'] - historical_zt['date']) == pd.Timedelta(days=1)
# 定义指标组合，计算概率
# 早封板: 首次封板时间 < '1030' (格式'HHMMSS'，取前4位)
historical_zt['early_seal'] = historical_zt['首次封板时间'].astype(str).str[:4] < '1030'
combos = {
'combo1': (historical_zt['换手率'] > 20) & (historical_zt['circ_mv'] < 50) & historical_zt['early_seal'], # 高换手+小市值+早封
'combo2': (historical_zt['换手率'] > 15) & historical_zt['所属行业'].str.contains('半导体|海南|机器人'), # 换手+热点（历史固定）
# 可添加更多
}
max_prob = 0
best_combo = None
for name, cond in combos.items():
subset = historical_zt[historical_zt['is_first_board'] & cond]
if len(subset) > 0:
prob = subset['next_day_zt'].mean()
if prob > max_prob:
max_prob = prob
best_combo = cond
print(f"最高概率组合: {max_prob*100:.2f}%")
return best_combo
# 辅助函数：回溯找到非空涨停数据日期（max_back_days=7）
def get_valid_zt_date(target_date, max_back_days=7):
date_obj = datetime.datetime.strptime(target_date, '%Y%m%d')
for i in range(max_back_days + 1):
check_date = (date_obj - datetime.timedelta(days=i)).strftime('%Y%m%d')
try:
df = ak.stock_zt_pool_em(date=check_date)
if not df.empty:
print(f"使用回溯日期: {check_date} (原: {target_date})")
return check_date, df
except:
pass
print("无可用涨停数据。")
return None, pd.DataFrame()
# 主函数（推荐时使用动态热点）
def recommend_stocks():
now = datetime.datetime.now()
today = now.strftime('%Y%m%d')
if now.hour >= 15:
# 分析今日首板，推荐明天买
analysis_date = today
buy_day = '明天'
else:
# 分析昨天首板，推荐今天买
yesterday = (now - datetime.timedelta(days=1)).strftime('%Y%m%d')
analysis_date = yesterday
buy_day = '今天'
# 获取分析日涨停池（回溯如果空）
analysis_date, zt_df = get_valid_zt_date(analysis_date)
if zt_df.empty:
print("无涨停数据，无法推荐。")
return
# 获取前日涨停池（回溯如果空）
prev_date = (datetime.datetime.strptime(analysis_date, '%Y%m%d') - datetime.timedelta(days=1)).strftime('%Y%m%d')
prev_date, prev_zt = get_valid_zt_date(prev_date)
# 处理空prev_zt
try:
prev_codes = prev_zt['代码']
except KeyError:
prev_codes = pd.Series()
# 过滤首板
first_board = zt_df[~zt_df['代码'].isin(prev_codes)]
if first_board.empty:
print("无首板股票，无法推荐。")
return
# 添加circ_mv
first_board['circ_mv'] = first_board['流通市值'] / 100000000
# 添加early_seal
first_board['early_seal'] = first_board['首次封板时间'].astype(str).str[:4] < '1030'
# 获取最佳组合（从历史）
best_combo = analyze_historical_prob()
# 为推荐动态更新热点（假设best_combo含热点；否则默认combo1）
hot_sectors = get_hot_sectors(analysis_date, top_n=10, sector_type="概念") # 或 "行业"
dynamic_contains = '|'.join(hot_sectors)
dynamic_combo = (first_board['换手率'] > 15) & first_board['所属行业'].str.contains(dynamic_contains)
if best_combo is None:
best_combo = (first_board['换手率'] > 20) & (first_board['circ_mv'] < 50) & first_board['early_seal']
else:
# 合并动态热点到best_combo
best_combo = dynamic_combo | best_combo # 或根据实际
# 应用最佳组合，推荐5只（按换手率降序）
recommended = first_board[best_combo].sort_values(by='换手率', ascending=False).head(5)
# 如果少于5，补充top换手率首板股
if len(recommended) < 5:
print(f"最高组合只找到{len(recommended)}只，补充top换手率首板股到5只。")
additional = first_board[~first_board['代码'].isin(recommended['代码'])].sort_values(by='换手率', ascending=False).head(5 - len(recommended))
recommended = pd.concat([recommended, additional])
print(f"基于{analysis_date}首板，推荐{buy_day}买的5只股票（最高概率组合，动态热点）：")
print(recommended[['代码', '名称', '换手率', 'circ_mv', '首次封板时间', '所属行业']])
if __name__ == '__main__':
new_var = recommend_stocks()
(由于只是测试版，后续还要根据市场完善，使用test临时命名)
通过网盘分享的文件：test_new.exe
链接: https://pan.baidu.com/s/1z2x0Qjg4i1Nwau98m2UFeQ?pwd=erv2 提取码: erv2
评论区好多告警的，或者是运行直接闪退的，问题优化
运行时长：10分钟
下载至桌面直接双击运行。运行中屏蔽告警，运行完毕后自动关闭并在桌面生成test.txt文件，可直接在txt文档中查看结果
上述如报错“”C:\\Users\\XXXXX\\AppData\\Local\\Temp\\_MEI25011\\akshare\\file_fold\\calendar.json'“”
可以试下，下方连接
链接:https://pan.baidu.com/s/1bvPbabQ46u7xQ4Rs4lC-Xg?pwd=3udt 提取码:3udt 复制这段内容后打开百度网盘手机App，操作更方便哦
有告警，需要根据运行图一图二使用命令行运行，运行时间10分钟
注：兴趣分享，仅供参考，不作为任何建议
补充运行方式：文件下载后直接放到桌面，win+R后按照图一，图二运行后就可以静等结果，不同配置电脑运行时间不同
| 
[复制链接]
发表于 2026-1-8 11:45
