[Python技术] 获取同花顺某个板块对应个股新增钉钉提醒

import os
import json
import requests
import pandas as pd
from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
# 配置参数
STOCK_BOARD_CODE = "xxxxx"
DINGDING_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=xxxxxx"
DATA_FILE = "stock_data.json"
def fetch_stocks(code):
    """获取指定板块的股票列表"""
    try:
        url = f"https://d.10jqka.com.cn/v2/blockrank/{code}/199112/d1000.js"
        headers = {
            'Referer': 'http://q.10jqka.com.cn/',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
                          '(KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36'
        }
        response = requests.get(url, headers=headers, timeout=10)
        if response.status_code == 200:
            # 处理JSONP数据
            json_str = response.text.split('(', 1)[1].rsplit(')', 1)[0]
            data = json.loads(json_str)
            #print(data)
            # 提取并展示数据
            stock_list = data.get('items', [])
            if stock_list:
                return pd.DataFrame(
                    [(s.get('5', '').zfill(6),
                      s.get('55', ''),
                      s.get('8', ''),
                      s.get('199112', 0))
                     for s in stock_list],
                    columns=['股票代码', '股票名称', '最新价', '涨跌幅']
                )
        return pd.DataFrame()  # 确保返回空DataFrame
    except Exception as e:
        print(f"获取股票数据失败: {e}")
        return pd.DataFrame()  # 确保返回空DataFrame
def save_stock_data(data):
    """保存股票数据到文件（接收DataFrame）"""
    # 转换为字典列表保存
    with open(DATA_FILE, "w", encoding="utf-8") as f:
        json.dump({
            "last_update": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "stocks": data.to_dict('records')  # 关键修复：DataFrame转字典列表
        }, f, ensure_ascii=False, indent=2)
def load_previous_data():
    """加载历史数据"""
    if not os.path.exists(DATA_FILE):
        return {"stocks": []}
    with open(DATA_FILE, "r", encoding="utf-8") as f:
        return json.load(f)
def compare_stocks(current_stocks, previous_stocks):
    """对比新旧股票数据（接收DataFrame）"""
    # 从历史数据中提取股票代码
    prev_codes = {s["股票代码"] for s in previous_stocks.get("stocks", [])}
    # 从当前DataFrame中提取新股票
    new_stocks = []
    for _, row in current_stocks.iterrows():
        if row['股票代码'] not in prev_codes:
            # 修改点：为股票名称添加超链接 [3,4](@ref)
            stock_name = row['股票名称']
            stock_code = row['股票代码']
            # 使用Markdown格式创建超链接 [1](@ref)
            hyperlinked_name = f"[{stock_name}](https://stockpage.10jqka.com.cn/{stock_code}/)"
            new_stocks.append(hyperlinked_name)
    return new_stocks
def send_dingding_alert(new_stocks, board_code):
    """通过钉钉发送新增股票提醒"""
    if not new_stocks:
        print("无新增股票")
        return
    message = f"🔔 **板块监控股票更新通知**\n\n" \
              f"**板块**: {board_code}\n" \
              f"**新增股票**:\n" + "\n".join(f"- {s}" for s in new_stocks)
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "title": "股票更新提醒",
            "text": message
        }
    }
    try:
        response = requests.post(
            DINGDING_WEBHOOK,
            headers={"Content-Type": "application/json"},
            data=json.dumps(payload)
        )
        print("钉钉通知发送状态:", response.status_code)
    except Exception as e:
        print(f"钉钉消息发送失败: {e}")
def scheduled_task():
    """定时任务主函数"""
    print(f"\n[{datetime.now()}] 开始执行定时任务")
    # 1. 获取历史数据
    previous_data = load_previous_data()
    # 2. 获取最新股票数据
    current_stocks = fetch_stocks(STOCK_BOARD_CODE)
    # 修复1：使用.empty判断DataFrame是否为空
    if current_stocks.empty:
        print("未获取到股票数据，任务终止")
        return
    # 3. 对比数据变化
    new_stocks = compare_stocks(current_stocks, previous_data)
    # 4. 发送钉钉提醒
    send_dingding_alert(new_stocks, STOCK_BOARD_CODE)
    # 5. 保存新数据
    save_stock_data(current_stocks)  # 修复2：直接传递DataFrame
    print("数据更新完成")
if __name__ == "__main__":
    # 创建调度器
    scheduler = BlockingScheduler()
    # 添加定时任务
    scheduler.add_job(
        scheduled_task,
        'cron',
        day_of_week='mon-fri',  # 周一至周五
        hour='15-23',
        minute='*',
    )
    print("定时任务已启动，等待执行...")
    try:
        scheduler.start()
    except KeyboardInterrupt:
        scheduler.shutdown()