Изучаем и парсим биржевую информацию Мосбиржи. Наш первый парсер на Python, разбор кода

Информационно-статистический сервер Московской Биржи (ИСС или ISS) – это сервис, предоставляющий разнообразную биржевую информацию в режиме реального времени, а также итоги торгов и статистические данные.

Основные возможности ИСС:

• Получение потоковых данных о ходе торгов.

• Просмотр и экспорт итогов торгов.

• Доступ к историческим данным по итогам торгов, ценам и прочим показателям.

• Выгрузка списков всех инструментов, режимы торгов и их группы.

• Мониторинг рыночной информации в различных разрезах.

  Данные о ходе торгов в режиме online и итоги торгов доступны только по подписке, естественно платной.

На сайте мосбиржи есть специальный раздел «Программный интерфейс к ИСС», на котором выложено Руководство разработчика (v.1.4), Описание метаданных и Описание методов.

С этих документов и надо начинать изучать ИИС. Кстати говоря Правила использования биржевой информации Московской Биржи четко определены и наглядно представлены в презентации.

Запросы к ИСС для получения информации формируются в виде URL с параметрами (или без), список всех доступных URL перечислен на странице Описание методов.

Получать информацию можно в форматах: XML, CSV, JSON, HTML

И давайте решим задачу по получению, к примеру:

• общей информации о рынках, содержащей блоки markets (рынки), boards (режимы торгов), board_groups (группировка режимов),

• рыночных параметров инструментов, содержащей блоки securities, marketdata, dataversion.

  Давайте перейдем на страницу  http://iss.moex.com/iss/index

красной рамкой выделен первый блок “engines”.

Давайте посмотрим эту страницу с расширением xml:

В ответах в форматах XML, HTML и JSON каждый блок информации отделен тегом data и содержит две секции: метаданные (описывают тип и длину полей с данными) и непосредственно рыночную информацию. Мы будем загонять эту информацию в табличный вид, где секция метаданные определит колонки, а секция с информацией заполнит ряды таблицы.

Ниже я предлагаю для внимания и использования листинг программы, благодаря которой можно спарсить информацию почти по любой ссылке на странице методов ISS Queries.

Общая идеология программы такая:

• Загрузка данных

• Парсинг XML

• Формирование DataFrame (таблицы)

• Сохранение в БД SQLite

• Сохранение (дублирование) в CSV файле с расширением txt.

  В программе используем библиотеки:

os – работа с файловой системой – создание директорий, получение путей к файлам и т.д.

sqlite3 – работа с базами данных SQLite – подключение, выполнение SQL-запросов, transaction и т.д.

xml.etree.ElementTree (ET) – парсинг и конвертация XML в древовидную структуру элементов для удобного доступа в коде Python.

pandas (pd) – работа с табличными данными, их обработка и анализ – DataFrame. Загрузка/запись данных в CSV и другие форматы.

requests – отправка HTTP запросов и получение ответов от веб-ресурсов. Используется для веб-скрейпинга и работы с API. Веб-скрейпинг – получения веб-данных путём извлечения их с веб-страниц. 

import os  # модуль для работы с файловой системой
import sqlite3  # модуль для работы с SQLite базами данных
import xml.etree.ElementTree as ET  # модуль для парсинга XML

import pandas as pd  # модуль для работы с данными
import requests  # модуль для HTTP запросов

url = "https://iss.moex.com/iss/index.xml"  # url источника данных XML - глобальные справочники ISS
# url = "https://iss.moex.com/iss/securities/SBER/indices.xml"  # url - Список индексов в которые входит бумага SBER.
# url = "https://iss.moex.com/iss/securities/IMOEX.xml"  # url - Получить спецификацию инструмента индекс Мосбиржи.

response = requests.get(url)  # запрос данных по url

root = ET.fromstring(response.content)  # парсинг XML из ответа

dfs = {}  # словарь для хранения dataframe

for data in root.findall("./data"):  # поиск всех элементов data
    id = data.attrib["id"]  # получение атрибута id

    columns = [
        c.attrib["name"] for c in data.findall("./metadata/columns/column")
    ]  # список колонок
    df = pd.DataFrame(columns=columns)  # создание пустого dataframe с определёнными колонками
    rows = data.findall("./rows/row")  # поиск строк

    for row in rows:
        data = dict(
            zip(columns, [row.attrib.get(c) for c in columns])
        )  # словарь данных строки
        df = pd.concat(
            [df, pd.DataFrame([data])], ignore_index=True
        )  # конкатенация строк

    dfs[id] = df  # сохранение dataframe в словаре

# далее код сохранения данных в БД и CSV
if not os.path.exists("moexmetadata"):
    os.mkdir("moexmetadata")

conn = sqlite3.connect("moexmetadata.db")

for id, df in dfs.items():
    df.to_sql(id, conn, if_exists="replace", index=False)
    # Проверка наличия колонки id в датафрейме
    if "id" in df.columns:
        cursor = conn.cursor()

        # Проверяем наличие индекса
        cursor.execute(f"PRAGMA index_info('index_name')")
        index_info = cursor.fetchall()

        if not index_info:
            # Создаем индекс в таблице по колонке id
            cursor.execute(f"CREATE INDEX index_name ON {id} (id)")
            conn.commit()

    # данные дублируем в файл
    filename = f"{id}.txt"
    path = os.path.join("moexmetadata", filename)
    df.to_csv(path, sep="\t", index=False)

conn.close()

Запускаем программу, она отрабатывает за несколько секунд. Можем в программе закомментировать первый url и раскоментировать второй, снова запустить, и т.д. при необходимости.

В каталоге с программой появилась папка “moexmetadata” с многочисленными файлами.

вот пример содержания

также у нас создалась БД с таблицами

и если совсем погружаться.., то теоретически можно между таблицами настроить логичные связи. Если конечно Вы сможете сначала разобраться во всем этом “хаосе” информации. Надо ли только это?

Идем далее

Согласно Руководства разработчика для некоторых запросов в конце файла также доступен специальный блок данных “cursor”.

например здесь http://iss.moex.com/iss/history/engines/stock/markets/shares/securities/MOEX он выглядит так

здесь INDEX – начало текущего набора данных в выдаче (номер строки), в запросе
задаётся параметром start; TOTAL – общий объём данных (количество строк), доступных по этому запросу;
PAGESIZE – объём текущего набора данных в выдаче (количество строк), в
запросе задаётся параметром limit (по умолчанию – 100, возможны значения 50,
20, 10, 5, 1).

Т.е. чтобы загрузить полный (очевидно, что большой) объём данных, нужно последовательно производить загрузку:
http://iss.moex.com/iss/history/engines/stock/markets/shares/securities/MOEX
http://iss.moex.com/iss/history/engines/stock/markets/shares/securities/MOEX?start=100
http://iss.moex.com/iss/history/engines/stock/markets/shares/securities/MOEX?start=200
и т.д.пока (INDEX + PAGESIZE) < TOTAL.

Кстати говоря также можно выгрузить все новости сайта мосбиржи со страницы https://iss.moex.com/iss/sitenews/ – порядка 43тысяч штук, только здесь каждый шаг будет содержать PAGESIZE=”50” записей.

Этот механизм вполне понятен и логичен, реализовать можно.

Но не все так однозначно, и как оказалось самый же первый URL на странице ISS Queries для получения списка всех бумаг торгуемых на московской бирже не имеет этого блока “cursor”. И очевидно, что на странице https://iss.moex.com/iss/securities выгружен далеко не полный список.

выгружать его надо также последовательно:

https://iss.moex.com/iss/securities.xml?start=0

https://iss.moex.com/iss/securities.xml?start=100

https://iss.moex.com/iss/securities.xml?start=200 и так далее. Только вот значения TOTAL у нас нет…, т.к. отсутствует раздел «cursor».

Уже после написания второй версии программы я обратил внимание, что в описании метода /iss/securities есть аргументы start и limit. По моему мнению необходимо было либо помещать раздел “cursor” везде, где подразумевается цикличный парсинг больших объемов – все как описано в руководстве, либо в руководстве предусмотреть все эти нюансы и упор делать на проверку наличия у метода аргументов start и limit.

Наверняка на мосбирже есть еще и другие подобные адреса с большим объемом данных и без указания общего объема TOTAL в блоке «cursor».

Итак, чтобы сильно не усложнять задачу для этого случая, я решил просто немного модифицировать программу. Мы принимаем, что нас интересует только один первый блок данных data (для таких больших страниц без блока «cursor» блок data будет единственным) и крутим цикл до тех пор, пока значение в не будет пустым.

Вот измененный код программы:

import os
import sqlite3
import xml.etree.ElementTree as ET

import pandas as pd
import requests

start = 0 # начальное значение
step = 100 # шаг парсинга 

while True:
    url = f"https://iss.moex.com/iss/securities.xml?start={start}"
    # url = f"https://iss.moex.com/iss/sitenews.xml?start={start}"
    response = requests.get(url)
    root = ET.fromstring(response.content)
    rows = root.findall("./data/rows/row")
    if not rows: # если нет строк - прерываем цикл
        break

    start += step
    print(start)
    dfs = {}  # словарь для хранения dataframe

    for data in root.findall("./data"):  # поиск всех элементов data
        id = data.attrib["id"]  # получение атрибута id
        columns = [
            c.attrib["name"] for c in data.findall("./metadata/columns/column")
        ]  # список колонок
        df = pd.DataFrame(columns=columns)  # создание пустого dataframe с определёнными колонками
        rows = data.findall("./rows/row")  # поиск строк

        for row in rows:
            data = dict(
                zip(columns, [row.attrib.get(c) for c in columns])
            )  # словарь данных строки
            df = pd.concat(
                [df, pd.DataFrame([data])], ignore_index=True
            )  # конкатенация строк

        dfs[id] = df  # сохранение dataframe в словаре

    # далее код сохранения данных в БД и CSV

    if not os.path.exists("moexmetadata"):
        os.mkdir("moexmetadata")

    conn = sqlite3.connect("moexmetadata.db")

    for id, df in dfs.items():
        df.to_sql(id, conn, if_exists="append", index=False)

        # Проверка наличия колонки id

        if "id" in df.columns:
            cursor = conn.cursor()

            # Проверяем наличие индекса

            cursor.execute(f"PRAGMA index_info('index_name')")

            index_info = cursor.fetchall()

            if not index_info:
                # если индекса нет, то создаем индекс в БД в таблице по колонке id

                cursor.execute(f"CREATE INDEX index_name ON {id} (id)")
                conn.commit()

        # данные дублируем в файл
        filename = f"{id}.txt"
        path = os.path.join("moexmetadata", filename)
        with open(path, "a", encoding="utf-8") as f:
            df.to_csv(f, sep="\t", index=False)

conn.close()

Запускаем, программа работает часа 3. В итоге получены внушительные данные

В базе данных «Список бумаг торгуемых на московской бирже» состоит из 595 тысяч записей.

На этом все. Первый результативный опыт получен. Есть общее понимание того, как устроена ISS MOEX. Получен опыт парсинга, get запросов, записи в файлы и БД.

А на очереди теперь у меня изучение алгопака Мосбиржи.

ВИДЕО по теме: