社区所有版块导航
Python
python开源   Django   Python   DjangoApp   pycharm  
DATA
docker   Elasticsearch  
aigc
aigc   chatgpt  
WEB开发
linux   MongoDB   Redis   DATABASE   NGINX   其他Web框架   web工具   zookeeper   tornado   NoSql   Bootstrap   js   peewee   Git   bottle   IE   MQ   Jquery  
机器学习
机器学习算法  
Python88.com
反馈   公告   社区推广  
产品
短视频  
印度
印度  
Py学习  »  Python

[Python技术] 通过yagmail发送邮件提醒

子晓聊技术 • 3 周前 • 75 次点击  

今天大盘表现不错, 买的个股20cm涨停, 虽然只吃了9cm多的肉,有点不想写技术文章了。 突然瞟了一眼,大盘个股涨停较多,想起有同学问我有没有发送邮件提醒的例子,那就写一个查询涨停发送邮件的功能。毕竟有些同学不需要钉钉、企业微信提醒,喜欢看邮件。
另外,最近好几个同学问怎么把代码部署到云服务器, 这里也推荐下我之前写的文章。
相关文章推荐:
【linux技术】阿里云轻量应用服务器通过宝塔linux面板安装软件流程记录
[linux技术] 新购aliyun服务器环境搭建基本步骤系列1
【Python技术】怎么借助同花顺问财搭建一套简易版的量化交易系统
【Python技术】每日最新消息发送到企业微信提醒

虽然涨停,但我深知追涨停不可取,用AI写一首劝诫远离追涨停的打油诗:

开盘涨停摆盛宴,
满仓杀入笑开颜。
封单如山夸海口,
尾盘泄洪泪涟涟。

追涨恰似扑灯蛾,

套牢方悟贪念多。

K线如猴窜天际,

账户浮沉荡秋千。

专家舌灿莲花绽,

连板妖股画饼圆。

七板狂欢半仓甩,

九成散户陷泥潭。

两油一涨全盘散,

三千绿韭泣荒原。

莫信暴富黄粱梦,

工资卡里见真仙。

忽闻邻户哭嚎声,
天台排队候神明。

清仓百万存银行,

方知稳字值千金。


最后附上源代码,需要的自取。 备注:如果发现格式有多余的特殊字符,用普通浏览器打开复制应该没问题。


备注下:我这里采用的163邮箱作为发送方, qq邮箱作为接收方。你完全可以改成自己的。 另外修改下定时任务的时间,我设置的收盘15:05执行定时任务。 写的这个样式有点low,暂时先这样。
import yagmailimport pywencaiimport pandas as pdimport scheduleimport timefrom datetime import datetime# ========== 配置部分 ==========EMAIL_USER = "youremail@163.com"  # 需开启SMTP服务EMAIL_PASSWORD = "yourpassword"  # 邮箱授权码(非登录密码)SMTP_SERVER = "smtp.163.com"  # 163邮箱服务器地址RECEIVERS = ["receiver@qq.com"]  # 收件人列表# ========== 数据获取函数 ==========def get_daily_limit_up():    """通过问财接口获取当日涨停数据(优化版)"""    try:        today = datetime.now().strftime('%Y%m%d')        query = f"{today}涨停,非ST,非科创板,非北交所"        df = pywencai.get(query=query, sort_key='连续涨停天数', sort_order='desc', loop=True)        return df[['股票代码''股票简称''最新价']].rename(            columns={'最新价''涨停价(元)'}        ) if not df.empty else pd.DataFrame()    except Exception as e:


    
        print(f"[{datetime.now()}] 数据获取失败: {str(e)}")        return pd.DataFrame()# ========== 邮件内容生成 ==========def generate_email_content(df):    """生成紧凑型邮件内容(带内联样式优化)"""    # 生成CSV附件(GBK编码兼容Excel)    csv_name = f"LimitUp_{datetime.now().strftime('%Y%m%d')}.csv"    df.to_csv(csv_name, index=False, encoding='gbk')    # 构建紧凑型HTML表格    html_table = df.to_html(        index=False,        classes=None,        border=0,        justify='center',        na_rep='-',        header=False    ).split('')[1].replace('''')    # 自定义紧凑表头    custom_thead = """                        股票代码            股票简称            涨停价(元)                """    # 内联样式优化(兼容各邮箱客户端)    html_content = f"""    
        font-family: 'Microsoft YaHei', Arial, sans-serif;        max-width: 800px;        margin: 0 auto;        padding: 0 !important;    ">       

            color: #2c3e50;

            border-bottom: 2px solid #3498db;            padding-bottom: 5px;            margin: 0 0 5px 0 !important;            font-size: 18px;        ">            {datetime.now().strftime('%Y-%m-%d')} 涨停统计(共 {len(df)} 只)                            width: 100%;            border-collapse: collapse;            margin: 0;            padding: 0;            font-size: 14px;        ">            {custom_thead}                            {html_table}                   
       

            color: #95a5a6;

            font-size: 12px;            margin: 10px 0 0 0;        ">            数据来源:同花顺问财 | 生成时间:{datetime.now().strftime('%H:%M:%S')}           
    """    return html_content, [csv_name]# ========== 邮件发送主逻辑 ==========def send_report():    """邮件发送核心逻辑(含异常重试)"""    try:        if datetime.now().weekday() >= 5:            print(f"[{datetime.now()}] 非交易日不发送")            return        df = get_daily_limit_up()        if df.empty:            print(f"[{datetime.now()}] 无有效涨停数据")            return        # 构建邮件内容        html_body, attachments = generate_email_content(df)        # 发送邮件(使用yagmail简化操作)        with yagmail.SMTP(user=EMAIL_USER, password=EMAIL_PASSWORD, host=SMTP_SERVER) as yag:            yag.send(                to=RECEIVERS,                subject=f"{datetime.now().strftime('%m%d')}涨停分析报告",                contents=html_body,                attachments=attachments            )        print(f"[{datetime.now()}] 邮件发送成功")    except Exception as e:         print(f"[{datetime.now()}] 发送失败: {str(e)}")        time.sleep(60)        send_report()# ========== 定时任务配置 ==========if __name__ == '__main__':    schedule.every().day.at("15:05").do(send_report)    print(f"[{datetime.now()}] 定时任务已启动...")    while True:        schedule.run_pending()        time.sleep(60)



如果对投资感兴趣,欢迎关注我这个号。 我时而去这个号胡扯下投资的事情,准备改个名(想了想,就叫 子晓聊投资, 名称还没申请下来)。为了监管风险,就分享一些投资上的心得。


Python社区是高质量的Python/Django开发社区
本文地址:http://www.python88.com/topic/182560
 
75 次点击