A股概念板块监控 + 邮件推送-平芜编程栈

脚本功能介绍

这是一个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`	发件邮箱 SMTP（QQ/163 等）
`FROM_ADDR`	发件邮箱
`AUTH_CODE`	授权码，支持环境变量`MAIL_AUTH_CODE`覆盖
`TO_ADDRS`	收件人列表，支持环境变量`MAIL_TO`（逗号分隔）覆盖
`TEST_MODE`	True = 只取前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_TO="a@163.com,b@qq.com" export MAIL_AUTH_CODE="新授权码" python3 concept_low_and_limit.py

九、局限与注意

非实时：日K接口有延迟，定位是收盘复盘
涨停判定简化：ST（5%）、北交所（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, retry=3, wait=1, **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(days=400)).strftime("%Y%m%d") return retry_call( ak.stock_board_concept_hist_em, symbol=name, period="daily", start_date=start, end_date=end, adjust="qfq" ) def is_new_low_1y(hist): if hist is None or len(hist) < 30: return False, None h = hist.sort_values("日期").reset_index(drop=True) 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, symbol=board) 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(days=45)).strftime("%Y%m%d") h = retry_call( ak.stock_zh_a_hist, symbol=code, period="daily", start_date=start, end_date=end, adjust="qfq" ) if h is None or len(h) == 0: return 0, None, None, None h = h.sort_values("日期").reset_index(drop=True) 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(days=30)).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_workers=6) 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_workers=8) 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(key=lambda x: (x["cnt"], x["last_day"] or ""), reverse=True) 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(key=lambda 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}" class="concept">{concept}<br><span class="sub">收盘 {first['concept_close']}</span></td> <td>{it['stock_code']}</td> <td class="stock">{name_disp}</td> <td class="price">{price_disp}</td> <td class="{pct_cls}">{pct_disp}</td> <td class="{cap_cls}">{cap_disp}</td> <td class="cnt">{it['limit_cnt']}</td> <td>{it['last_limit_day']}</td> </tr>""" else: body_rows += f""" <tr> <td>{it['stock_code']}</td> <td class="stock">{name_disp}</td> <td class="price">{price_disp}</td> <td class="{pct_cls}">{pct_disp}</td> <td class="{cap_cls}">{cap_disp}</td> <td class="cnt">{it['limit_cnt']}</td> <td>{it['last_limit_day']}</td> </tr>""" html = f""" <!DOCTYPE html> <html> <head> <meta charset="utf-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 class="container"> <div class="header"> <h1>📉 概念板块创新低监控报告</h1> <p>生成时间：{now}</p> </div> <div class="summary"> <div class="card"><div class="num">{low_count}</div><div class="label">1年内创新低概念</div></div> <div class="card"><div class="num">{len(set(r['concept'] for r in rows))}</div><div class="label">含涨停股的概念</div></div> <div class="card"><div class="num">{len(rows)}</div><div class="label">上榜强势股(前3)</div></div> <div class="card"><div class="num">{sum(1 for r in rows if r.get('big_cap'))}</div><div class="label">⭐ 流通市值≥500亿</div></div> </div> <div class="legend">图例：⭐ + 紫色市值 = 流通市值 ≥ {BIG_CAP_THRESHOLD} 亿；红色涨幅 = 上涨，绿色 = 下跌</div> <table> <thead> <tr> <th style="width:18%">概念板块</th> <th style="width:10%">股票代码</th> <th style="width:16%">股票名称</th> <th style="width:10%">目前股价</th> <th style="width:10%">今日涨幅</th> <th style="width:11%">流通市值(亿)</th> <th style="width:11%">近1月涨停次数</th> <th style="width:14%">最近涨停日</th> </tr> </thead> <tbody> {body_rows} </tbody> </table> <div class="footer"> <p class="warn">⚠ 本报告由程序自动生成，数据来自公开接口，仅供参考，不构成投资建议。</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()