В прошлой статье я довольно подробно разобрал пример парсинга поисковых запросов 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)
