Python爬取多个时间段资金净流入排名靠前清单

  import akshare as akimport pandas as pdimport matplotlib.pyplot as pltimport smtplibfrom email.mime.text import MIMETextfrom email.mime.multipart import MIMEMultipartfrom email.mime.image import MIMEImageimport timeimport os# =========================================================# 1. 邮箱配置（必须修改）# =========================================================SENDER_EMAIL = "你的QQ号@qq.com"SENDER_PASSWORD = "你的QQ邮箱授权码"   # 不是QQ密码！RECEIVER_EMAIL = "接收者邮箱@xx.com"SMTP_SERVER = "smtp.qq.com"SMTP_PORT = 465# =========================================================# 2. 获取不同周期的资金流向前10# =========================================================PERIODS = {    "即时": "即时",    "3日": "3日排行",    "5日": "5日排行",    "10日": "10日排行",    "20日": "20日排行",    "60日": "60日排行",}def get_top10_by_period(period_name):    """获取指定周期的资金流向前10"""    print(f"📊 正在获取【{period_name}】资金流向数据...")    try:        df = ak.stock_fund_flow_individual(symbol=PERIODS[period_name])        # 自动识别列名        name_col = None        net_col = None        for col in df.columns:            if "简称" in str(col) or "名称" in str(col):                name_col = col            if "净额" in str(col):                net_col = col        if not name_col or not net_col:            print(f"❌ 列名识别失败，当前列: {df.columns.tolist()}")            return None        # 单位转换        def convert(val):            val = str(val).strip()            try:                if "亿" in val:                    return float(val.replace("亿", ""))                elif "万" in val:                    return float(val.replace("万", "")) / 10000                else:                    return float(re.sub(r"[^\d.-]", "", val)) / 100000000            except:                return 0        df["净额数值"] = df[net_col].apply(convert)        top10 = df.sort_values("净额数值", ascending=False).head(10)        return top10[[name_col, net_col, "净额数值"]]    except Exception as e:        print(f"❌ 获取【{period_name}】失败: {e}")        return None# =========================================================# 3. 绘制柱状图# =========================================================def draw_bar_chart(df, period_name):    """绘制资金流向柱状图"""    try:        plt.rcParams["font.sans-serif"] = ["SimHei"]        plt.rcParams["axes.unicode_minus"] = False        fig, ax = plt.subplots(figsize=(12, 6))        names = df.iloc[:, 0].tolist()        values = df["净额数值"].tolist()        bars = ax.bar(names, values, color="#4C72B0")        ax.set_title(f"{period_name} 主力资金净流入 Top 10", fontsize=16)        ax.set_ylabel("净流入金额（亿元）", fontsize=12)        ax.tick_params(axis="x", rotation=45, labelsize=10)        # 标注数值        for bar in bars:            height = bar.get_height()            ax.annotate(                f"{height:.2f}",                xy=(bar.get_x() + bar.get_width() / 2, height),                xytext=(0, 3),                textcoords="offset points",                ha="center",                va="bottom",                fontsize=9,            )        plt.tight_layout()        img_path = f"fund_flow_{period_name}.png"        plt.savefig(img_path, dpi=150, bbox_inches="tight")        plt.close(fig)        print(f"✅ 【{period_name}】图表已生成")        return img_path    except Exception as e:        print(f"❌ 绘制【{period_name}】图表失败: {e}")        return None# =========================================================# 4. 发送汇总邮件# =========================================================def send_summary_email(image_paths):    """发送包含所有图表的邮件"""    print(f"📧 正在发送汇总邮件至 {RECEIVER_EMAIL} ...")    msg = MIMEMultipart()    msg["From"] = SENDER_EMAIL    msg["To"] = RECEIVER_EMAIL    msg["Subject"] = f"📈 个股资金流向汇总 - {time.strftime('%Y-%m-%d')}"    body = MIMEText(        f"""        <html>        <body>            <h2>沪深两市主力资金净流入排名</h2>            <p>以下是今日不同周期的主力资金净流入前10名股票统计：</p>            <ul>                <li>即时</li>                <li>3日排行</li>                <li>5日排行</li>                <li>10日排行</li>                <li>20日排行</li>                <li>60日排行</li>            </ul>            <p>详细图表请见附件。</p>        </body>        </html>        """,        "html",        "utf-8",    )    msg.attach(body)    # 添加所有图片附件    for img_path in image_paths:        if img_path and os.path.exists(img_path):            with open(img_path, "rb") as f:                img_data = f.read()            image = MIMEImage(img_data, name=os.path.basename(img_path))            image.add_header(                "Content-Disposition",                "attachment",                filename=os.path.basename(img_path),            )            msg.attach(image)    try:        with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) as server:            server.login(SENDER_EMAIL, SENDER_PASSWORD)            server.send_message(msg)        print("🎉 汇总邮件发送成功！")    except Exception as e:        print(f"❌ 邮件发送失败: {e}")# =========================================================# 5. 主程序# =========================================================if __name__ == "__main__":    print("=" * 60)    print("📈 开始获取个股资金流向数据")    print("=" * 60)    image_paths = []    for period_name in PERIODS.keys():        df = get_top10_by_period(period_name)        if df is not None and not df.empty:            print(f"\n📋 【{period_name}】Top 5 预览：")            print(df.head().to_string(index=False))            img_path = draw_bar_chart(df, period_name)            if img_path:                image_paths.append(img_path)        else:            print(f"⚠️ 【{period_name}】数据为空，跳过")        time.sleep(2)  # 防止请求过快    if image_paths:        send_summary_email(image_paths)    else:        print("❌ 未生成任何图表，不发送邮件")    print("\n✅ 所有任务执行完毕")