脚本功能介绍这是一个A股概念板块监控 邮件推送自动化脚本核心功能是找出1年内创新低的概念板块统计这些板块中近1个月涨停次数最多的股票生成美化 HTML 报告并发送到指定邮箱一、核心流程5 步text① 拉取全部概念板块 ↓ ② 筛选1年内创新低的概念 ↓ ③ 加载全市场市值快照 ↓ ④ 统计创新低概念中成分股的近1月涨停次数取每概念前3 ↓ ⑤ 生成美化 HTML 邮件发送到多个收件人二、逐项功能说明1. 概念板块全量获取通过ak.stock_board_concept_name_em()拉取全部概念板块数百个支持测试模式TEST_MODE True时只取前 10 个快速验证链路2. 1年内创新低判定对每个概念板块拉取近 400 天日K前复权判定逻辑最新收盘价 ≤ 近 250 个交易日最低收盘价输出创新低概念列表 各自最新收盘点位3. 全市场市值快照批量提速用ak.stock_zh_a_spot_em()一次性拉全市场行情构建{股票代码: 流通市值(亿)}映射避免逐只查市值接口速度大幅提升缓存复用全流程只拉一次4. 个股涨停统计对创新低概念的每只成分股拉取近 45 天日K统计最近 30 个自然日内的涨停次数主板涨跌幅 ≥ 9.5%创业板/科创板300/301/688/689≥ 19.5%同时获取最新收盘价、今日涨幅、流通市值每个概念按涨停次数降序取前 3 只5. 邮件美化与发送生成 HTML 邮件含顶部渐变标题 生成时间4 个摘要卡片创新低概念数、含涨停股概念数、上榜股票数、⭐500亿市值股数图例说明分组表格同一概念的股票合并概念单元格底部风险提示支持多收件人逐个发送失败重试指数退避收件人之间互不可见三、报告字段字段说明概念板块创新低的概念名称 板块收盘点位股票代码6 位代码股票名称名称⭐ 前缀表示流通市值 ≥ 500 亿目前股价前复权最新收盘价今日涨幅最近交易日涨跌幅红涨绿跌带/-流通市值(亿)≥500 亿的紫色加粗标记近1月涨停次数近 30 自然日涨停次数最近涨停日最近一次涨停日期四、配置项配置说明SMTP_HOST / SMTP_PORT发件邮箱 SMTPQQ/163 等FROM_ADDR发件邮箱AUTH_CODE授权码支持环境变量MAIL_AUTH_CODE覆盖TO_ADDRS收件人列表支持环境变量MAIL_TO逗号分隔覆盖TEST_MODETrue 只取前10概念快速验证MAIL_RETRY单封邮件失败重试次数BIG_CAP_THRESHOLD大市值阈值默认 500 亿五、技术特性特性实现并发拉取板块筛选 6 线程、个股统计 8 线程失败重试数据接口 3 次重试邮件 2 次重试 退避缓存优化市值快照全局缓存只拉一次多收件人逐个单独发送隐私好、互不影响美化输出HTML 内联 CSS渐变标题、卡片、分组表格、红涨绿跌、标记高亮进度可见终端逐概念打印完成进度容错单只股票/单个板块失败不影响整体停牌处理无数据显示-不参与标记六、适用场景收盘复盘每个交易日收盘后跑一次邮件推送弱势板块中的强势股超跌反弹观察创新低板块 有涨停股可能是资金试探性介入大市值关注⭐ 标记帮你在小盘异动中快速识别大市值标的定时任务配合 cron 每个交易日 15:30 自动运行七、数据口径说明项目口径复权前复权qfq创新低最新收盘 ≤ 近250交易日最低收盘涨停按代码前缀区分 9.5% / 19.5% 阈值ST 5%、北交所 30% 未单独处理今日涨幅盘中延迟快照收盘后收盘涨幅市值流通市值元转亿元八、使用方式# 安装依赖 pip3 install --upgrade akshare pandas # 测试模式快速验证 sed -i s/TEST_MODE False/TEST_MODE True/ concept_low_and_limit.py python3 concept_low_and_limit.py # 全量运行 sed -i s/TEST_MODE True/TEST_MODE False/ concept_low_and_limit.py python3 concept_low_and_limit.py # 多收件人 环境变量授权码 export MAIL_TOa163.com,bqq.com export MAIL_AUTH_CODE新授权码 python3 concept_low_and_limit.py九、局限与注意非实时日K接口有延迟定位是收盘复盘涨停判定简化ST5%、北交所30%未单独处理可能误判耗时全量约 5~15 分钟取决于概念数量和接口速度接口变动AKShare 字段可能随版本变化报错先升级邮件限制QQ 邮箱日发信量有限收件人别太多不构成投资建议创新低≠买入涨停多≠必涨十、脚本AUTH_CODE os.getenv(MAIL_AUTH_CODE, 授权码换自己的)FROM_ADDR 自己的发信箱TO_ADDRS 列表里邮箱改成自己的收件箱#!/usr/bin/env python3 # -*- coding: utf-8 -*- import akshare as ak import pandas as pd import smtplib import os import time import traceback from email.mime.text import MIMEText from email.header import Header from email.utils import formataddr from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta # 配置 SMTP_HOST smtp.qq.com SMTP_PORT 465 FROM_ADDR 发信箱qq.com AUTH_CODE os.getenv(MAIL_AUTH_CODE, sreztnoovatsbada) _env_to os.getenv(MAIL_TO, ) TO_ADDRS [a.strip() for a in _env_to.split(,) if a.strip()] or [ 收件箱163.com, ] TEST_MODE False MAIL_RETRY 2 BIG_CAP_THRESHOLD 500 # 亿元 # 工具 def retry_call(func, *args, retry3, wait1, **kwargs): for i in range(retry): try: return func(*args, **kwargs) except Exception: if i retry - 1: return None time.sleep(wait) return None # 全市场市值快照一次性 _CAP_CACHE None def get_market_cap_map(): 返回 {code: 流通市值(亿)}一次性拉取失败返回空字典 global _CAP_CACHE if _CAP_CACHE is not None: return _CAP_CACHE df retry_call(ak.stock_zh_a_spot_em) if df is None or df.empty: _CAP_CACHE {} return _CAP_CACHE cap_map {} for _, r in df.iterrows(): code str(r.get(代码, )).zfill(6) cap r.get(流通市值) try: cap_map[code] round(float(cap) / 1e8, 2) if pd.notna(cap) else None except Exception: cap_map[code] None _CAP_CACHE cap_map return cap_map # 1. 全部概念 def get_all_concepts(): df retry_call(ak.stock_board_concept_name_em) if df is None: return [] names df[板块名称].tolist() if TEST_MODE: names names[:10] return names # 2. 板块历史判断1年新低 def board_hist(name): end datetime.now().strftime(%Y%m%d) start (datetime.now() - timedelta(days400)).strftime(%Y%m%d) return retry_call( ak.stock_board_concept_hist_em, symbolname, perioddaily, start_datestart, end_dateend, adjustqfq ) def is_new_low_1y(hist): if hist is None or len(hist) 30: return False, None h hist.sort_values(日期).reset_index(dropTrue) close h[收盘].astype(float) latest_close close.iloc[-1] window close.tail(250) min_close window.min() return latest_close min_close 1e-8, round(float(latest_close), 2) # 3. 板块成分股 def get_cons(board): df retry_call(ak.stock_board_concept_cons_em, symbolboard) return df if df is not None else pd.DataFrame() # 4. 个股近1月涨停次数 最新价 今日涨幅 def stock_limit_count(code): end datetime.now().strftime(%Y%m%d) start (datetime.now() - timedelta(days45)).strftime(%Y%m%d) h retry_call( ak.stock_zh_a_hist, symbolcode, perioddaily, start_datestart, end_dateend, adjustqfq ) if h is None or len(h) 0: return 0, None, None, None h h.sort_values(日期).reset_index(dropTrue) latest h.iloc[-1] latest_close round(float(latest[收盘]), 2) today_pct round(float(latest[涨跌幅]), 2) if 涨跌幅 in h.columns else None cutoff (datetime.now() - timedelta(days30)).strftime(%Y-%m-%d) h2 h[h[日期].astype(str) cutoff] if h2.empty: return 0, None, latest_close, today_pct pct h2[涨跌幅].astype(float) is_20 code.startswith((300, 301, 688, 689)) threshold 19.5 if is_20 else 9.5 limit_days h2[pct threshold] cnt len(limit_days) last_day limit_days[日期].iloc[-1] if cnt 0 else None return cnt, (str(last_day) if last_day is not None else None), latest_close, today_pct # 5. 主流程 def main(): print(① 拉取全部概念板块...) concepts get_all_concepts() print(f 共 {len(concepts)} 个概念 (测试模式 if TEST_MODE else )) print(② 筛选1年内创新低的概念...) low_concepts [] def worker_board(name): h board_hist(name) ok, close is_new_low_1y(h) return name, ok, close with ThreadPoolExecutor(max_workers6) as ex: futs {ex.submit(worker_board, n): n for n in concepts} for f in as_completed(futs): try: name, ok, close f.result() if ok: low_concepts.append((name, close)) except Exception: pass print(f 创新低概念{len(low_concepts)} 个) if not low_concepts: print( 无创新低概念退出) return print(③ 加载全市场市值快照...) cap_map get_market_cap_map() print(f 市值记录{len(cap_map)} 条) print(④ 统计这些概念成分股近1月涨停次数...) result [] for idx, (cname, cclose) in enumerate(low_concepts, 1): cons get_cons(cname) if cons.empty: continue stock_rows [] def worker_stock(row): code str(row.get(代码, )).zfill(6) name str(row.get(名称, )) cnt, last_day, price, today_pct stock_limit_count(code) cap cap_map.get(code) return {code: code, name: name, cnt: cnt, last_day: last_day, price: price, today_pct: today_pct, cap: cap} with ThreadPoolExecutor(max_workers8) as ex: futs [ex.submit(worker_stock, row) for _, row in cons.iterrows()] for f in as_completed(futs): try: stock_rows.append(f.result()) except Exception: pass stock_rows.sort(keylambda x: (x[cnt], x[last_day] or ), reverseTrue) top3 [s for s in stock_rows if s[cnt] 0][:3] for s in top3: result.append({ concept: cname, concept_close: cclose, stock_code: s[code], stock_name: s[name], price: s[price], today_pct: s[today_pct], cap: s[cap], big_cap: (s[cap] is not None and s[cap] BIG_CAP_THRESHOLD), limit_cnt: s[cnt], last_limit_day: s[last_day] or -, }) print(f [{idx}/{len(low_concepts)}] {cname} 完成) if not result: print( 无涨停股票退出) return result.sort(keylambda x: (x[concept], -x[limit_cnt])) print(⑤ 生成美化邮件并发送...) html build_html(result, len(low_concepts)) ok send_mail(【股市监控】概念创新低 近1月涨停统计, html) print( 邮件发送 (成功 if ok else 失败)) # 6. 美化 HTML def build_html(rows, low_count): now datetime.now().strftime(%Y-%m-%d %H:%M:%S) from collections import OrderedDict grouped OrderedDict() for r in rows: grouped.setdefault(r[concept], []).append(r) body_rows for concept, items in grouped.items(): first items[0] span len(items) for i, it in enumerate(items): price_disp f{it[price]:.2f} if it.get(price) is not None else - tp it.get(today_pct) if tp is None: pct_disp, pct_cls -, else: pct_disp f{tp:.2f}% pct_cls up if tp 0 else (down if tp 0 else ) cap it.get(cap) if cap is None: cap_disp, cap_cls -, else: cap_disp f{cap:.0f} cap_cls bigcap if it.get(big_cap) else name_disp it[stock_name] if it.get(big_cap): name_disp f⭐ {name_disp} if i 0: body_rows f tr td rowspan{span} classconcept{concept}brspan classsub收盘 {first[concept_close]}/span/td td{it[stock_code]}/td td classstock{name_disp}/td td classprice{price_disp}/td td class{pct_cls}{pct_disp}/td td class{cap_cls}{cap_disp}/td td classcnt{it[limit_cnt]}/td td{it[last_limit_day]}/td /tr else: body_rows f tr td{it[stock_code]}/td td classstock{name_disp}/td td classprice{price_disp}/td td class{pct_cls}{pct_disp}/td td class{cap_cls}{cap_disp}/td td classcnt{it[limit_cnt]}/td td{it[last_limit_day]}/td /tr html f !DOCTYPE html html head meta charsetutf-8 style body {{ font-family: Microsoft YaHei, Arial, sans-serif; background:#f5f7fa; margin:0; padding:20px; }} .container {{ max-width: 1120px; margin:0 auto; background:#fff; border-radius:10px; overflow:hidden; box-shadow:0 2px 10px rgba(0,0,0,.08); }} .header {{ background:linear-gradient(135deg,#1e3c72,#2a5298); color:#fff; padding:24px 30px; }} .header h1 {{ margin:0; font-size:22px; }} .header p {{ margin:6px 0 0; font-size:13px; opacity:.85; }} .summary {{ display:flex; gap:20px; padding:18px 30px; background:#f0f4ff; border-bottom:1px solid #e0e6f0; flex-wrap:wrap; }} .card {{ flex:1; min-width:150px; background:#fff; border-radius:8px; padding:14px 18px; border-left:4px solid #2a5298; }} .card .num {{ font-size:22px; font-weight:bold; color:#1e3c72; }} .card .label {{ font-size:12px; color:#666; margin-top:4px; }} table {{ width:100%; border-collapse:collapse; font-size:13px; }} thead th {{ background:#1e3c72; color:#fff; padding:10px 8px; text-align:left; font-weight:500; }} tbody td {{ padding:9px 8px; border-bottom:1px solid #eef1f6; vertical-align:top; }} tbody tr:hover {{ background:#f7f9ff; }} .concept {{ font-weight:bold; color:#1e3c72; background:#f4f7ff; }} .sub {{ font-weight:normal; color:#888; font-size:11px; }} .stock {{ color:#c0392b; font-weight:500; }} .price {{ color:#2c3e50; font-weight:500; }} .up {{ color:#c0392b; font-weight:500; }} .down {{ color:#27ae60; font-weight:500; }} .bigcap {{ color:#8e44ad; font-weight:bold; background:#f5eef8; text-align:center; }} .cnt {{ color:#d35400; font-weight:bold; text-align:center; }} .footer {{ padding:16px 30px; font-size:12px; color:#999; text-align:center; border-top:1px solid #eee; }} .warn {{ color:#c0392b; }} .legend {{ padding:10px 30px; font-size:12px; color:#666; background:#fafbfd; }} /style /head body div classcontainer div classheader h1 概念板块创新低监控报告/h1 p生成时间{now}/p /div div classsummary div classcarddiv classnum{low_count}/divdiv classlabel1年内创新低概念/div/div div classcarddiv classnum{len(set(r[concept] for r in rows))}/divdiv classlabel含涨停股的概念/div/div div classcarddiv classnum{len(rows)}/divdiv classlabel上榜强势股(前3)/div/div div classcarddiv classnum{sum(1 for r in rows if r.get(big_cap))}/divdiv classlabel⭐ 流通市值≥500亿/div/div /div div classlegend图例⭐ 紫色市值 流通市值 ≥ {BIG_CAP_THRESHOLD} 亿红色涨幅 上涨绿色 下跌/div table thead tr th stylewidth:18%概念板块/th th stylewidth:10%股票代码/th th stylewidth:16%股票名称/th th stylewidth:10%目前股价/th th stylewidth:10%今日涨幅/th th stylewidth:11%流通市值(亿)/th th stylewidth:11%近1月涨停次数/th th stylewidth:14%最近涨停日/th /tr /thead tbody {body_rows} /tbody /table div classfooter p classwarn⚠ 本报告由程序自动生成数据来自公开接口仅供参考不构成投资建议。/p p创新低 ≠ 买入信号涨停次数多 ≠ 一定上涨请结合基本面与量能判断。/p /div /div /body /html return html # 7. 发送邮件多收件人 重试 def send_mail(subject, html_content): success, failed 0, [] for addr in TO_ADDRS: sent False last_err None for attempt in range(1, MAIL_RETRY 2): try: msg MIMEText(html_content, html, utf-8) msg[From] formataddr((股市监控, FROM_ADDR)) msg[To] addr msg[Subject] Header(subject, utf-8) with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT) as server: server.login(FROM_ADDR, AUTH_CODE) server.sendmail(FROM_ADDR, [addr], msg.as_string()) sent True break except Exception as e: last_err e if attempt MAIL_RETRY: time.sleep(2 * attempt) if sent: success 1 print(f ✅ 已发送: {addr}) else: failed.append(addr) print(f ❌ 发送失败: {addr} - {last_err}) time.sleep(0.5) print(f 发送完成成功 {success} 个失败 {len(failed)} 个) if failed: print(f 失败列表: {failed}) return success 0 if __name__ __main__: main()十一、效果截图
