Національна фондова біржа Індії (National Stock Exchange of India Limited) (NSE) є однією з провідних фондових бірж Індії, розташованою в Мумбаї.
Ця стаття зосереджена на витягуванні щоденних даних для індексів NSE за допомогою Python, зокрема на отриманні даних на рівні індексів, а не для окремих акцій, що складають ці індекси.
На кінець 2024 року NSE відслідковує близько 120 індексів.
Дотримуючись наведених нижче кроків, ми отримаємо історичні щоденні дані для багатьох ринкових індексів NSE та збережемо їх як одноразовий дамп даних у вказану папку.
Крок 1: Завантаження файлу Market Watch-Indicies (.csv)
Перейдіть на сайт NSE: https://www.nseindia.com/market-data/live-market-indices.
Натисніть кнопку ‘Download (.csv)’ у верхньому правому куті, щоб завантажити Excel файл, що містить назви індексів. Ми будемо використовувати ці назви індексів для ітерації в нашому коді.
https://www.nseindia.com/market-data/live-market-indices.
Крок 2: Використання бібліотеки nselib для отримання даних
Ми будемо використовувати бібліотеку nselib
(версія 1.5) для отримання даних.
# Встановлення необхідних пакетів (виконати цю команду, якщо ще не встановлено)
# ! pip -q install nselib
# Імпортування необхідних бібліотек
import os
import warnings
import numpy as np
import datetime as dt
import pandas as pd
from nselib import capital_market
# Приховати попередження та налаштувати опції Pandas
warnings.filterwarnings('ignore')
pd.set_option('mode.chained_assignment', None)
# Перевірка версії nselib
import nselib
print("nselib version:", nselib.__version__)
# Перевірка: Отримання даних для Nifty 50
data = capital_market.index_data(index='NIFTY 50',
from_date='01-01-2015',
to_date='01-01-2020',
period=None)
display(data)
Функція capital_market.index_data()
приймає чотири параметри для витягування щоденних даних: назву індексу (із файлу CSV), дати початку/закінчення (у форматі DD-MM-YYYY) та період (встановлено в None).
Крок 3: Написання функції для ітерації по всіх індексах
Ми почнемо з визначення поточної дати, яка слугуватиме як кінцева дата для отримання даних.
# Визначення поточної дати як кінцевої дати
final_end_date = dt.date.today()
print("Дата закінчення витягування даних:", final_end_date)
>> Дата закінчення витягування даних: 2024-12-28
Далі ми зчитаємо завантажений файл, щоб витягти та зберегти назви індексів для ітерації.
# Читання файлу введення та витягування назв індексів
input_df = pd.read_csv('MW-All-Indices-28-Dec-2024.csv')
input_df.columns = input_df.columns.str.strip()
index_names = input_df['INDEX'].tolist()
print("Кількість індексів:", len(index_names))
>> Кількість індексів: 121
Тепер ми напишемо функцію для отримання всіх історичних даних для індексів NSE. Ця функція буде:
- Отримувати щоденні ціни закриття за допомогою
capital_market.index_data()
. - Фільтрувати дані, щоб включити лише дату, назву індексу та ціну закриття.
- Форматувати дані для подальшого використання.
def get_full_history_nse_py(index_name, final_start_date, final_end_date):
"""
Отримати щоденні історичні дані про ціни закриття для заданого індексу.
Параметри:
index_name (str): Назва індексу.
final_start_date (str): Початкова дата для отримання даних.
final_end_date (str): Кінцева дата для отримання даних.
Повертає:
pd.DataFrame: Фільтровані та відформатовані дані, що містять дату, ціну закриття та назву індексу.
"""
# Отримання даних з NSE
df = capital_market.index_data(
index=index_name,
from_date=final_start_date,
to_date=final_end_date,
period=None
)
# Вибірка та перейменування стовпців
df = df[['TIMESTAMP', 'CLOSE_INDEX_VAL', 'INDEX_NAME']]
df.rename(columns={'TIMESTAMP': 'Date',
'CLOSE_INDEX_VAL': 'Close',
'INDEX_NAME': 'Index'}, inplace=True)
# Форматування даних
df['Date'] = pd.to_datetime(df['Date'], format='%d-%m-%Y')
df = df.round({'Close': 1})
df.drop_duplicates(subset=['Date'], keep='first', inplace=True)
return df
# Тестування функції
get_full_history_nse_py('NIFTY 50', '01-01-2024', '28-12-2024')
Нарешті, ми проходимо через всі назви індексів, щоб отримати та зберегти їх історичні дані.
# Визначення стартового року та поточного року
START_YEAR = 1995
curr_year = dt.datetime.today().year
# Перетворення кінцевої дати на формат рядка
final_end_date = final_end_date.strftime('%d-%m-%Y')
# Ініціалізація лічильника для оброблених індексів
iterator = 0
# Ітерація по назвам індексів та отримання даних
for index in index_names:
for year in range(START_YEAR, curr_year + 1):
# Визначення початкової дати
final_start_date = dt.date(year, 1, 1).strftime('%d-%m-%Y')
try:
# Отримання історичних даних
df = get_full_history_nse_py(index, final_start_date, final_end_date)
# Форматування та збереження даних
df['Date'] = df['Date'].dt.strftime('%d-%m-%Y')
df = df.dropna(how='all').drop_duplicates()
df.to_csv(f"{index}_{final_start_date}_{final_end_date}.csv", index=False)
# Інкремент лічильника та логування прогресу
iterator += 1
print(f"Оброблений індекс:{index} | Кількість файлів:{iterator}/{len(index_names)}")
# Очищення та вихід з внутрішнього циклу
del df
break
except Exception as e:
# Продовжити, якщо виникає помилка
continue
Бібліотека успішно обробила 72 з 121 індексів, що є розумним результатом, враховуючи зусилля, необхідні для ручного оброблення кожного індексу. Код виконався за 88 хвилин (~1.5 години) на моїй локальній системі, що демонструє, що цей підхід пропонує масштабовану методику для отримання даних індексів NSE.
Крок 4: Об’єднання всіх кроків
Наступний код об’єднує всі попередні кроки для отримання та збереження щоденних історичних даних для індексів NSE, починаючи з 1995 року (залежно від доступності для кожного індексу). Дані кожного індексу зберігаються у файлі CSV для подальшого аналізу.
# Встановлення необхідних пакетів (виконати цю команду, якщо ще не встановлено)
# ! pip -q install nselib
# Імпортування необхідних бібліотек
import os
import warnings
import numpy as np
import datetime as dt
import pandas as pd
from nselib import capital_market
# Приховати попередження та налаштувати опції Pandas
warnings.filterwarnings('ignore')
pd.set_option('mode.chained_assignment', None)
# Перевірка версії nselib
import nselib
print("nselib version:", nselib.__version__)
print()
# Визначення поточної дати як кінцевої дати
final_end_date = dt.date.today()
print("Дата закінчення витягування даних:", final_end_date)
print()
# Читання файлу введення та витягування назв індексів
input_df = pd.read_csv('MW-All-Indices-28-Dec-2024.csv')
input_df.columns = input_df.columns.str.strip()
index_names = input_df['INDEX'].tolist()
print("Кількість індексів:", len(index_names))
print()
# Визначення функції
def get_full_history_nse_py(index_name, final_start_date, final_end_date):
"""
Отримати щоденні історичні дані про ціни закриття для заданого індексу.
Параметри:
index_name (str): Назва індексу.
final_start_date (str): Початкова дата для отримання даних.
final_end_date (str): Кінцева дата для отримання даних.
Повертає:
pd.DataFrame: Фільтровані та відформатовані дані, що містять дату, ціну закриття та назву індексу.
"""
# Отримання даних з NSE
df = capital_market.index_data(
index=index_name,
from_date=final_start_date,
to_date=final_end_date,
period=None
)
# Вибірка та перейменування стовпців
df = df[['TIMESTAMP', 'CLOSE_INDEX_VAL', 'INDEX_NAME']]
df.rename(columns={'TIMESTAMP': 'Date',
'CLOSE_INDEX_VAL': 'Close',
'INDEX_NAME': 'Index'}, inplace=True)
# Форматування даних
df['Date'] = pd.to_datetime(df['Date'], format='%d-%m-%Y')
df = df.round({'Close': 1})
df.drop_duplicates(subset=['Date'], keep='first', inplace=True)
return df
# Визначення стартового року та поточного року
START_YEAR = 1995
curr_year = dt.datetime.today().year
# Перетворення кінцевої дати на формат рядка
final_end_date = final_end_date.strftime('%d-%m-%Y')
# Ініціалізація лічильника для оброблених індексів
iterator = 0
# Ітерація по назвам індексів та отримання даних
for index in index_names:
for year in range(START_YEAR, curr_year + 1):
# Визначення початкової дати
final_start_date = dt.date(year, 1, 1).strftime('%d-%m-%Y')
try:
# Отримання історичних даних
df = get_full_history_nse_py(index, final_start_date, final_end_date)
# Форматування та збереження даних
df['Date'] = df['Date'].dt.strftime('%d-%m-%Y')
df = df.dropna(how='all').drop_duplicates()
df.to_csv(f"{index}_{final_start_date}_{final_end_date}.csv", index=False)
# Інкремент лічильника та логування прогресу
iterator += 1
print(f"Оброблений індекс:{index} | Кількість файлів:{iterator}/{len(index_names)}")
# Очищення та вихід з внутрішнього циклу
del df
break
except Exception as e:
# Продовжити, якщо виникає помилка
continue
Дивлячись вперед
Для періодичних оновлень, розгляньте можливість реалізації наступних покращень:
- Перевірка наявності існуючих даних: Перевірка, чи дані для конкретного індексу вже були отримані та збережені в призначеній папці.
- Отримання інкрементальних оновлень: Визначення останньої дати оновлення та отримання лише нових даних, починаючи з останньої дати оновлення до найостаннішої доступної дати.
Сподіваюся, цей посібник буде корисним. Не соромтеся поділитися своїми відгуками!
Перекладено з: Fetching NSE Indices Data Using Python