使用python的akshare库读取股票实时数据并保存

Posted onAuthorcnliutz

这段代码的主要功能是定时获取 A 股股票数据并保存到 Excel 文件中，同时记录操作日志。具体步骤如下：

  1. 导入必要的库：导入 akshare、pandas、os、schedule、time 和 logging 库，用于获取股票数据、数据处理、文件路径操作、定时任务、时间处理和日志记录。
  2. 定义获取并保存股票数据的函数：get_and_save_stock_data 函数从东方财富获取最新的 A 股股票数据，生成带有时间戳的文件名，将数据保存为 Excel 文件，并记录操作成功或失败的日志。
  3. 配置日志记录：使用 logging.basicConfig 配置日志记录，设置日志级别为 INFO，并指定日志格式。
  4. 设置定时任务：使用 schedule 库设置定时任务，每 10 分钟调用一次 get_and_save_stock_data 函数。
  5. 循环检查并执行定时任务：使用 while True 循环不断检查是否有定时任务到期，并执行到期的任务。每次循环暂停 1 秒，以避免过度占用 CPU 资源。

通过以上步骤，代码实现了定时获取和保存股票数据的功能，并记录了操作日志。
在这个修改后的代码中，我们使用 try-except 块包裹了 while 循环，当用户按下 Ctrl+C 时，会触发 KeyboardInterrupt 异常，程序会捕获这个异常并记录一条日志信息，然后正常退出。

import akshare as ak
import pandas as pd
import os
import schedule
import time
import logging

def get_and_save_stock_data():
        """
        Fetches the latest A-share stock data from Eastmoney and saves it to an Excel file.
        The file is named with a timestamp and stored in the same directory as the script.
        Logs the success or failure of the operation.
        """
        try:
            # 获取股票数据
            data = ak.stock_zh_a_spot_em()
            print(data.head())
            # 生成带时间的文件名
            current_time = time.strftime("%Y%m%d%H%M%S")
            # Generate a file name with a timestamp
            file_name = f'stock_data_{current_time}.xlsx'
            # Create a full file path by combining the directory of the current script and the file name
            file_path = os.path.join(os.path.dirname(__file__), file_name)
            # Write the data to an Excel file, without including the index
            data.to_excel(file_path, index=False)
            # Log the successful retrieval and storage of the stock data
            logging.info(f"Successfully retrieved stock data and saved to {file_path} at {time.strftime('%Y-%m-%d %H:%M:%S')}")
        except Exception as e:
            # Log any errors that occur during the retrieval or storage of the stock data
            logging.error(f"An error occurred while retrieving or saving stock data: {e}")

# Configure logging to record events at the INFO level
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Schedule the function to run every 10 or 0.5 minutes
schedule.every(0.5).minutes.do(get_and_save_stock_data)


# Start the scheduling loop，Ctrl+C to stop 终止运行
try:
    # Continuously check and run scheduled tasks
    while True:
        # Check if any scheduled tasks are due and run them
        schedule.run_pending()
        # Pause the loop for 1 second to avoid excessive CPU usage
        time.sleep(1)
except KeyboardInterrupt:
    logging.info("Program terminated by user.结束")