数据抓取与时间操作：结合 datetime 与 timedelta 进行定时任务管理

原创

jackcode

发布于 2024-10-29 12:11:25

12000

代码可运行

文章被收录于专栏：爬虫资料爬虫资料

运行总次数：0

代码可运行

在现代数据分析中，数据的实时性和准确性尤为重要，尤其是金融数据，如股票信息。本文将详细探讨如何利用 datetime 和 timedelta 库来管理数据抓取的定时任务。通过定时触发数据采集任务，我们可以实时获取纳斯达克股市的开盘数据。同时，为了提高爬虫的采集效率与稳定性，本文还将结合代理 IP 技术、多线程技术以及其他一些优化手段进行详细介绍。

1. 定时任务管理概述

Python 的 datetime 和 timedelta 是两个处理时间的基础库，其中 datetime 可以创建和管理日期、时间对象，而 timedelta 则提供了灵活的时间增量功能。借助这两个库，可以有效地管理数据抓取的时间任务，确保每次任务在预定时间内触发。

在金融数据采集中，定时任务主要用来实现以下几点功能：

定期抓取：在纳斯达克股市开盘时间内定期抓取数据。
时段控制：确保数据只在市场开盘时间内采集。
避免频繁请求：防止对服务器造成压力，也减少封 IP 风险。

在 Python 中可以使用 datetime 获取当前时间，并使用 timedelta 设置下一次任务触发时间，实现自动化的定时数据采集任务。

2. 数据抓取细节

为了实现对纳斯达克股市的实时数据采集，我们将实现以下功能：

代理 IP：为了减少 IP 被封的风险，我们将使用第三方代理 IP 服务，比如爬虫代理。
User-Agent 与 Cookie：自定义请求头，增加请求的模拟性。
多线程：使用 ThreadPoolExecutor 来实现多线程，提高抓取效率。

代码实现

以下是一个简单的实现示例，包括定时任务管理、代理设置、数据请求及多线程处理的完整代码。

import requests
import datetime
import time
import threading
from concurrent.futures import ThreadPoolExecutor
from datetime import timedelta

# 代理信息 (以16yun“爬虫代理加强版”为例)
PROXY_HOST = "proxy.16yun.cn"
PROXY_PORT = "12345"  # 示例端口
PROXY_USER = "your_username"
PROXY_PASS = "your_password"

# 设置代理
proxy_meta = f"http://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}"
proxies = {
    "http": proxy_meta,
    "https": proxy_meta,
}

# 请求头信息
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.45 Safari/537.36",
    "Cookie": "your_cookie_here"  # 替换成实际的 Cookie
}

# 纳斯达克开盘时间（美东时间9:30 - 16:00）
OPEN_TIME = datetime.time(9, 30)
CLOSE_TIME = datetime.time(16, 0)

# 数据抓取函数
def fetch_stock_data():
    try:
        url = "https://example.com/nasdaq_stock_data"  # 替换为实际的纳斯达克数据网址
        response = requests.get(url, headers=headers, proxies=proxies, timeout=5)
        if response.status_code == 200:
            data = response.json()
            # 处理数据
            print(f"数据抓取成功：{data}")
        else:
            print(f"请求失败，状态码：{response.status_code}")
    except requests.RequestException as e:
        print(f"请求异常：{e}")

# 判断是否在开盘时间
def is_market_open():
    now = datetime.datetime.now().time()
    return OPEN_TIME <= now <= CLOSE_TIME

# 定时任务管理器
def schedule_task(interval=60):
    while True:
        if is_market_open():
            # 使用多线程提高效率
            with ThreadPoolExecutor(max_workers=5) as executor:
                futures = [executor.submit(fetch_stock_data) for _ in range(5)]
                for future in futures:
                    try:
                        future.result()
                    except Exception as e:
                        print(f"线程异常：{e}")
            print(f"下一次数据采集将在 {interval} 秒后执行...")
        else:
            print("市场未开盘，等待开盘...")
        time.sleep(interval)

# 开始定时任务
if __name__ == "__main__":
    # 每隔60秒采集一次数据
    schedule_task(interval=60)

代码说明

代理 IP 设置：代码中的代理设置使用了爬虫代理代理的配置格式，并将代理信息集成在 proxies 参数中，使每次请求都能通过代理发送。
定时任务与时间检查：is_market_open 函数检查当前时间是否在纳斯达克开盘时段。主函数 schedule_task 会定期检查时间，并在开盘期间执行数据采集任务。
多线程数据抓取：使用 ThreadPoolExecutor 执行多线程抓取任务，模拟多个并发请求。这不仅能提高抓取效率，还能增加获取数据的成功率。
异常处理：通过 try-except 捕获请求中的异常，防止程序在网络故障时中断。