如何用Python一键获取连板天梯数据——自动化复盘的开始

# 安装tusharepip install tushare

import tushare as ts# 设置token（替换成你自己的）ts.set_token('你的token')pro = ts.pro_api()# 获取指定日期的连板天梯df = pro.limit_step(trade_date='20260715')print(df.head(10))

import tushare as tsimport pandas as pdfrom datetime import datetime, timedeltats.set_token('你的token')pro = ts.pro_api()# 获取最近5个交易日的连板天梯def get_limit_step_multiple(days=5):# 这里需要结合交易日历，简化版用日期递减 end_date = datetime.now().strftime('%Y%m%d') all_data = [] for i in range(days): date = (datetime.now() - timedelta(days=i)).strftime('%Y%m%d') try: df = pro.limit_step(trade_date=date) if not df.empty: all_data.append(df) except: pass if all_data: return pd.concat(all_data, ignore_index=True) return pd.DataFrame()df = get_limit_step_multiple(5)print(f”获取到{len(df)}条连板记录”)

pip install akshare

import akshare as akimport pandas as pd# 获取指定日期的涨停股池df = ak.stock_zt_pool_em(date='20260715')print(df.head(10))

import akshare as akdef get_limit_up_board(date): ”””获取连板天梯数据””” df = ak.stock_zt_pool_em(date=date) if df.empty: return pd.DataFrame()# 筛选连板股（连板数>=2） df['连板数'] = df['连板数'].astype(int) df_limit = df[df['连板数'] >= 2].copy()# 按连板数降序排列 df_limit = df_limit.sort_values('连板数', ascending=False)# 只保留关键字段 result = df_limit[['名称', '代码', '最新价', '涨跌幅', '换手率', '连板数',  '首次涨停时间', '最终涨停时间', '封单金额']] return result# 使用示例df = get_limit_up_board('20260715')print(f”今日连板股数量：{len(df)}”)print(df)

import akshare as ak# 获取昨日涨停今日表现df_yesterday = ak.stock_yesterday_zt_em(date='20260715')print(df_yesterday.head())

import akshare as akimport pandas as pdfrom datetime import datetime# 获取A股全历史交易日trade_date_range = ak.tool_trade_date_hist_sina()trade_date_range['trade_date'] = pd.to_datetime(trade_date_range['trade_date']).dt.date# 判断今天是否为交易日today = datetime.now().date()is_trade = today in trade_date_range['trade_date'].valuesprint(f”今天是否为交易日：{is_trade}”)# 输出：True 或 False

pip install levistock

import levistock as lk# 获取涨停板股票池df = lk.stock_zt_pool_em()print(df.head())# 返回涨停股列表

import levistock as lk# 获取市场情绪数据data = lk.market_emotion_cls()print(f”市场热度: {data['market_degree']}”)print(f”涨停家数: {data['up_down_dis']['up_num']}”)# 输出涨停家数

import akshare as akimport pandas as pdfrom datetime import datetimedef daily_limit_report(date=None): ”””生成每日连板复盘报告””” if date is None: date = datetime.now().strftime('%Y%m%d') print(f”========== {date} 连板复盘报告 ==========\n”)# 1. 获取涨停股池 df = ak.stock_zt_pool_em(date=date) if df.empty: print(”今日无涨停数据（可能为非交易日）”) return# 2. 统计基础数据 total_limit = len(df) df['连板数'] = df['连板数'].astype(int) limit_2plus = len(df[df['连板数'] >= 2]) max_boards = df['连板数'].max() print(f”📊 涨停总数：{total_limit}家”) print(f”📈 连板股数（≥2板）：{limit_2plus}家”) print(f”🏆 最高连板：{max_boards}板\n”)# 3. 连板天梯 print(”【连板天梯】”) for boards in range(max_boards, 0, -1): count = len(df[df['连板数'] == boards]) bar = '█' * min(count, 20) print(f” {boards}板：{count}家 {bar}”)# 4. 高连板个股明细 print(”\n【高连板个股（≥3板）】”) high = df[df['连板数'] >= 3].sort_values('连板数', ascending=False) if not high.empty: for _, row in high.iterrows(): print(f” {row['名称']}（{row['代码']}）：{row['连板数']}板 | ” f”换手{row['换手率']}% | 封单{row['封单金额']:.0f}万”) else: print(” 无3板以上个股”)# 5. 保存数据 df.to_csv(f'limit_up_{date}.csv', encoding='utf-8-sig') print(f”\n✅ 数据已保存至：limit_up_{date}.csv”)# 执行daily_limit_report()

========== 20260715 连板复盘报告 ==========📊 涨停总数：89家📈 连板股数（≥2板）：24家🏆 最高连板：7板【连板天梯】 7板：1家 █ 6板：2家 ██ 5板：3家 ███ 4板：5家 █████ 3板：6家 ██████ 2板：7家 ███████【高连板个股（≥3板）】 粤桂股份（000833.SZ）：7板 | 换手8.5% | 封单24500万 东方精工（002611.SZ）：6板 | 换手12.3% | 封单18300万 ...