Scraping (парсинг) сайта HistData.com на Python c Google Colab

В прошлой статье я довольно подробно разобрал пример парсинга поисковых запросов Google и статей Bloomberg. Для обучения нейронок данных нужно много. 🙂

В этой статье рассмотрю как выполнить парсинг слегка защищенных сайтов, вроде histdata.com Попиарю этот ресурс — он очень хорош для анализа данных. Непросто найти бесплатно архивные данные по курсам валют и финансовым индексам с интервалом в 1 минуту. Ресурс шикарный и 30 USD за доступ — он того стоит, но проблема в том, что похоже, ресурс работает на автомате.

Мои попытки что-то уточнить у администраци используя контактные данным успехом не увенчались. Оплачивать за доступ к услугам с риском, что если что-то пойдет не так, то ответа от техподдержки не будет — как-то не очень приятно. В общем, я рекомендую использовать легальный способ получения данных, но если риски высокие не получить, то можно спарсить.

В общем, цель — забрать файлы данных (zip архивы) с разных страниц сайта. Прямых линков на архивы нет. Загрузка происходит JavaScript разбираться с которым нет никакого желания. Поэтому проще всего сэмулировать поведение белкового пользователя, т.е. открыть страничку и кликнуть по нужной ссылке. Предварительно эмулятор браузера подготавлявается так, чтобы не возникало никаких диалогов и файлы сохранялись в нужной папке.

Загрузка Python библиотек для scraping данных

#@title Код импорта библиотек
%%capture

!pip install selenium
!apt-get update # to update ubuntu to correctly run apt install
!apt install chromium-chromedriver
!cp /usr/lib/chromium-browser/chromedriver /usr/bin
import sys
sys.path.insert(0,'/usr/lib/chromium-browser/chromedriver')
from selenium import webdriver
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

import numpy as np
from bs4 import BeautifulSoup
import string
import time
from datetime import datetime
from google.colab import files

!pip install fake-useragent
from fake_useragent import UserAgent
import os

Загрузка Selenium драйвера Chrome

Здесь несколько моментов уточню. Python-овский код ниже подготавливает Chrome драйвер, чтобысайт думал, что по нему ходит реальный пользователь. При этом в настройках браузера в опциях отключаются все пользовательские диалоги, т.е. он переводится в silent режим. Кроме того указывается путь для скачивания, но далее этот путь будет менятся другим кодом.

#@title Код загрузки драйвера Chrome
download_dir = "Downloads"
def loadChrome(download_dir = "Downloads"):
  chrome_options = webdriver.ChromeOptions()
  chrome_options.add_argument('--headless')
  chrome_options.add_argument('--no-sandbox')
  chrome_options.add_argument('--disable-dev-shm-usage')
  chrome_options.add_argument("--window-size=1920x1080")
  chrome_options.add_argument("--disable-notifications")
  chrome_options.add_argument('--verbose')
  if not os.path.isdir(download_dir):
    os.mkdir(download_dir)
  chrome_options.add_experimental_option("prefs", {
      "profile.default_content_settings.popups": False, 
      "download.default_directory": download_dir, #"C://Temp", #"C:\\Temp", #<path_to_download_default_directory>
      "download.prompt_for_download": False,
      "download.directory_upgrade": True,
      "safebrowsing_for_trusted_sources_enabled": False,
      "safebrowsing.enabled": False
  })
  chrome_options.add_argument('--disable-gpu')
  chrome_options.add_argument('--disable-software-rasterizer')

  ua = UserAgent()
  user_agent = "user-agent=" + ua.random
  print("User-agent:", user_agent)
  chrome_options.add_argument(user_agent)

  browser = webdriver.Chrome('chromedriver', chrome_options = chrome_options)
  return browser

browser = loadChrome()

Получение списка доступных файлов с HistData

Нужно забрать по некоторому URL с сайта указывающему на страницу с нужными финансовыми индексами список всех доступных для скачивания файлов. Нам для тренировки нейронной сети нужны были данные минимум за 3 года, т.ена каждый индекс 3 файла по году + 6 файлов помесячно за текущий год. Здесь особо никаких тонкостей нет. Скачиваются страница с индесом и забираются с помощью BeautifulSoup ссылки на страницы с файлами.

#Download any web page
def getWebPage(browser, url):
  browser.get(url)
  html_source = browser.page_source
  soup = BeautifulSoup(html_source, 'lxml')
  return html_source, soup
#@title Получаем список для скачивания
#Получаем список ссылок для чкачки
def getHistDataDownloadLinks(browser, url):
  matches = ['2020', '2019', '2018', '2017']
  html_source, soup = getWebPage(browser, url) 
  symbol = url[-6:]
  histdata_links = soup.find_all('div', {'class' : 'page-content'})
  data = []
  if ((histdata_links != None) and (len(histdata_links)) > 0):
    for link in histdata_links[0].find_all('a'):
      if any(x in link['href'] for x in matches):
        data.append('https://www.histdata.com' + link['href'])
  return data, symbol

Скачивание файлов с HistData эмулируя работу пользователя

Функция прописывающая путь для скачивания файлов драйвером Chromium.

#@title Enable browser headless
def enable_download_headless(browser, download_dir):
    browser.command_executor._commands["send_command"] = ("POST", '/session/$sessionId/chromium/send_command')
    params = {'cmd':'Page.setDownloadBehavior', 'params': {'behavior': 'allow', 'downloadPath': download_dir}}
    browser.execute("send_command", params)

def createDir(browser, symbol):
  path = "/content/Downloads/" + symbol
  if (not os.path.isdir(path)):
    os.mkdir(path)
  enable_download_headless(browser, path)
  return path

Поскольку скачиваемые файлы запакованы в zip, нужна функция для распаковки. Мне нужно взять 9 файлов *.csv, склеить в один большой файл и затем запаковать в zip, чтобы скачивание с FTP сервера в Colab занимало как можно меньше времени. Скачивание — процесс не быстрый, распаковка гораздо шустрее просходит.

#@title Zip/UnZip file
import zipfile
def unzip(path_to_zip_file, directory_to_extract_to):
  if (os.path.isfile(path_to_zip_file)):
    with zipfile.ZipFile(path_to_zip_file, 'r') as zip_ref:
      print("Unzip file:", path_to_zip_file)
      zip_ref.extractall(directory_to_extract_to)
      return True
  else:
    print("File is not found:", path_to_zip_file)    
  return False    

def zip(path_to_file, zip_file_name):
  if (os.path.isfile(zip_file_name)):
    print("Remove file:", zip_file_name)
    os.remove(zip_file_name)

  with zipfile.ZipFile(zip_file_name, "w", zipfile.ZIP_DEFLATED) as zip_ref:
    print("Compress file:", path_to_file, "to file", zip_file_name)
    zip_ref.write(path_to_file, os.path.basename(path_to_file))  

Поскольку разработчики сайта позаботились о защите от автоматического скачивания, прямой ссылки нет. Чтобы скачать файл нужно кликнуть на ссылку мышью. Чтобы сэмулировать клик мыши:

, затем

  • Элемент находится на странице по id ‘a_file’.
  • Виртуальный экран скроллируется до этого элемента, чтобы он был в визуально доступной области запуском скрипта «arguments[0].scrollIntoView();».
  • Используется метод click() для скачивания файла в дефолтную директорию указанную в свойствах драйвера Chrome.
#@title Download HistData File
def downloadHistDataFile(browser, symbol):
  path = createDir(browser, symbol)
  # function to handle setting up headless download
  #path = download_dir # + "/" + symbol  
  if (not os.path.isdir(path)):
    os.makedirs(path)
  enable_download_headless(browser, path)
  for link in data:
    if ("javascript:history.back" not in link):
      browser.get(link)
      element = browser.find_element_by_id('a_file')
      browser.execute_script("arguments[0].scrollIntoView();", element) #Scroll to element, otherwise error
      element.click()
  return path  

Объединение CSV файлов с HistData

После скачивания файлов необходимо:

объединить их

  • Распаковать zip файлы c *.csv данными.
  • Извлеченные *.csv файлы склеить в один большой файл для удобства анализа.
  • Запаковать в zip архив.
  • Скачать с Colab на локальный ПК.

В общем-то никаких нюансов при реализации алгоритма нет, поэтому не буду останавливаться.

#@title Merge CSV file. Zip its. Download.
#from os import listdir
def mergeCSVfiles(path, symbol, prefix = 'HISTDATA_COM_ASCII_'):
  order = ['2017', '2018', '2019', '2020']
  cmd = ""

  os.chdir(path + "/")
  print(os.getcwd())

  for date in order:
    if '2020' in date:
      today = datetime.today()
      for month in range(1, today.month + 1):
        month = str(month)
        if len(month) == 1:
          month = '0'+ str(month) 
        file = prefix + symbol + "_M12020" + month + ".zip" 
        print("Try to unzip file:", file)
        if unzip(file, path + "/"):
          cmd += ' DAT_ASCII_' + symbol + "_M1_2020" + month + ".csv"
    else:
      file = prefix + symbol + "_M1" + date + ".zip"
      #full_path = file #path + "/" + 
      print("Try to unzip file:", file)
      if unzip(file, path + "/"):
        cmd += ' DAT_ASCII_' + symbol + '_M1_' + date + ".csv"     
 
  newfile = "DAT_ASCII_" + symbol + "_M1.csv"
  zipfile = "DAT_ASCII_" + symbol + "_M1.zip"

  if (os.path.isfile(newfile)):
    print("Delete file:", newfile)
    os.remove(newfile)

  print(os.getcwd())
  cmd = "cat " + cmd.strip() + " > " + newfile #+ "/content/" + path
  
  print(cmd)
  os.system(cmd)

  zip(newfile, zipfile)
  files.download(zipfile)

  return zipfile

Ну и запуск разработанного кода. В symbols передается строковое имя нужного символа и далее код

symbols = "USDCHF"
data, symbol = getHistDataDownloadLinks(browser, 'https://www.histdata.com/download-free-forex-historical-data/?/ascii/1-minute-bar-quotes/' + symbols)
print(np.array(data))
path = downloadHistDataFile(browser, symbol)
print(path)
#Иногда требуется задержка, поскольку Colab распараллеливает скачивание и к моменту запуска mergeCSVfiles иногда не все файлы закачались. 
mergeCSVfiles(path, symbol)
Spread the love
Запись опубликована в рубрике IT рецепты с метками . Добавьте в закладки постоянную ссылку.

Обсуждение закрыто.