You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

循环遍历股票列表存数据遇问题及3个月数据爬取需求

NSE股票数据爬取问题解决方案

问题概述

需从NSE官网获取约500只股票数据,操作流程为:

  1. 打开目标URL
  2. 输入股票代码并选择下拉框首个选项
  3. 点击「1W」查看周数据,提取指定日期的Total Traded Quantity、No. of Trades、Deliverable Qty、% Dly Qt to Traded Qty字段
  4. 数据保存至CSV

现有代码存在两个问题:

  • 输入框清除不彻底,导致后续股票数据重复
  • 缺少点击「3M」获取3个月完整数据的功能

解决方案

1. 修复输入框重复数据问题

原代码的clear()方法可能因页面交互逻辑失效,改用全选后删除+JavaScript清空的组合方式确保输入框完全清空;同时保留单次浏览器实例,避免重复启动Chrome。

2. 新增3M数据获取功能

单独实现点击「3M」按钮的逻辑,并提取该时间段内的所有数据行(而非指定日期),适配批量数据导出需求。


完整修正代码

版本1:指定日期(1W模式)

import undetected_chromedriver as uc
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.keys import Keys
from bs4 import BeautifulSoup
import csv
import os

# 配置项
stock_symbols = ["INFY", "TCS", "RELIANCE", "HDFCBANK", "ITC"]  # 替换为500只股票代码
target_date = "11-Apr-2025"  # 每日修改此处日期
csv_file = "nse_weekly_data.csv"
wait_timeout = 15

# 初始化浏览器(仅启动一次)
driver = uc.Chrome()
driver.get("https://www.nseindia.com/report-detail/eq_security")
driver.maximize_window()
wait = WebDriverWait(driver, wait_timeout)

def get_weekly_stock_data(symbol):
    # 彻底清空输入框
    symbol_input = wait.until(EC.element_to_be_clickable((By.ID, "hsa-symbol")))
    symbol_input.click()
    # 全选后删除 + JS清空双保险
    symbol_input.send_keys(Keys.CONTROL + "a")
    symbol_input.send_keys(Keys.DELETE)
    driver.execute_script("arguments[0].value = '';", symbol_input)
    
    # 输入新股票代码
    symbol_input.send_keys(symbol)
    print(f"✅ 输入股票代码: {symbol}")
    
    # 等待下拉框加载并选择首个选项
    wait.until(EC.visibility_of_element_located((By.ID, "hsa-symbol_listbox")))
    wait.until(EC.element_to_be_clickable((By.XPATH, "//div[@id='hsa-symbol_listbox']//div[1]"))).click()
    print(f"✅ 选择股票: {symbol}")
    
    # 点击1W按钮
    wait.until(EC.element_to_be_clickable((By.ID, "oneW"))).click()
    print(f"✅ 切换至周数据视图")
    
    # 等待表格加载完成
    wait.until(EC.visibility_of_element_located((By.ID, "hsaTable")))
    soup = BeautifulSoup(driver.page_source, "html.parser")
    table = soup.find("table", {"id": "hsaTable"})
    rows = table.find_all("tr")
    
    # 映射目标列索引
    headers = [th.text.strip() for th in rows[0].find_all("th")]
    cols_to_extract = ["Date", "Total Traded Quantity", "No. of Trades", "Deliverable Qty", "% Dly Qt to Traded Qty"]
    col_indices = {col: headers.index(col) for col in cols_to_extract}
    
    # 查找目标日期数据
    for row in rows[1:]:
        cols = [td.text.strip() for td in row.find_all("td")]
        if cols and cols[col_indices["Date"]] == target_date:
            return {
                "Symbol": symbol,
                "Date": cols[col_indices["Date"]],
                "Total Traded Quantity": cols[col_indices["Total Traded Quantity"]],
                "No. of Trades": cols[col_indices["No. of Trades"]],
                "Deliverable Qty": cols[col_indices["Deliverable Qty"]],
                "% Dly Qt to Traded Qty": cols[col_indices["% Dly Qt to Traded Qty"]],
            }
    
    print(f"❌ {symbol} 未找到目标日期数据")
    return None

# 写入CSV
file_exists = os.path.isfile(csv_file)
for symbol in stock_symbols:
    data = get_weekly_stock_data(symbol)
    if data:
        with open(csv_file, mode="a", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=data.keys())
            if not file_exists:
                writer.writeheader()
                file_exists = True  # 避免重复写入表头
            writer.writerow(data)
        print(f"✅ {symbol} 数据已写入CSV")

driver.quit()
print("✅ 所有周数据提取完成")

版本2:3个月完整数据模式

import undetected_chromedriver as uc
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.keys import Keys
from bs4 import BeautifulSoup
import csv
import os

# 配置项
stock_symbols = ["INFY", "TCS", "RELIANCE", "HDFCBANK", "ITC"]  # 替换为500只股票代码
csv_file = "nse_3month_data.csv"
wait_timeout = 15

# 初始化浏览器(仅启动一次)
driver = uc.Chrome()
driver.get("https://www.nseindia.com/report-detail/eq_security")
driver.maximize_window()
wait = WebDriverWait(driver, wait_timeout)

def get_3month_stock_data(symbol):
    # 彻底清空输入框
    symbol_input = wait.until(EC.element_to_be_clickable((By.ID, "hsa-symbol")))
    symbol_input.click()
    symbol_input.send_keys(Keys.CONTROL + "a")
    symbol_input.send_keys(Keys.DELETE)
    driver.execute_script("arguments[0].value = '';", symbol_input)
    
    # 输入股票代码并选择
    symbol_input.send_keys(symbol)
    print(f"✅ 输入股票代码: {symbol}")
    wait.until(EC.visibility_of_element_located((By.ID, "hsa-symbol_listbox")))
    wait.until(EC.element_to_be_clickable((By.XPATH, "//div[@id='hsa-symbol_listbox']//div[1]"))).click()
    print(f"✅ 选择股票: {symbol}")
    
    # 点击3M按钮
    wait.until(EC.element_to_be_clickable((By.ID, "threeM"))).click()
    print(f"✅ 切换至3个月数据视图")
    
    # 等待表格加载
    wait.until(EC.visibility_of_element_located((By.ID, "hsaTable")))
    soup = BeautifulSoup(driver.page_source, "html.parser")
    table = soup.find("table", {"id": "hsaTable"})
    rows = table.find_all("tr")
    
    # 映射列索引
    headers = [th.text.strip() for th in rows[0].find_all("th")]
    cols_to_extract = ["Date", "Total Traded Quantity", "No. of Trades", "Deliverable Qty", "% Dly Qt to Traded Qty"]
    col_indices = {col: headers.index(col) for col in cols_to_extract}
    
    # 提取所有行数据
    all_data = []
    for row in rows[1:]:
        cols = [td.text.strip() for td in row.find_all("td")]
        if cols:
            all_data.append({
                "Symbol": symbol,
                "Date": cols[col_indices["Date"]],
                "Total Traded Quantity": cols[col_indices["Total Traded Quantity"]],
                "No. of Trades": cols[col_indices["No. of Trades"]],
                "Deliverable Qty": cols[col_indices["Deliverable Qty"]],
                "% Dly Qt to Traded Qty": cols[col_indices["% Dly Qt to Traded Qty"]],
            })
    
    print(f"✅ {symbol} 提取到 {len(all_data)} 条3个月数据")
    return all_data

# 写入CSV
file_exists = os.path.isfile(csv_file)
for symbol in stock_symbols:
    data_list = get_3month_stock_data(symbol)
    if data_list:
        with open(csv_file, mode="a", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=data_list[0].keys())
            if not file_exists:
                writer.writeheader()
                file_exists = True
            writer.writerows(data_list)
        print(f"✅ {symbol} 3个月数据已写入CSV")

driver.quit()
print("✅ 所有3个月数据提取完成")

使用说明

  1. 替换stock_symbols列表为你的500只股票代码
  2. 每日执行时,在周数据版本中修改target_date参数即可
  3. 两个版本独立运行,分别生成对应CSV文件

内容的提问来源于stack exchange,提问作者chintan patel

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.13 09:22:34