Блог им. imagic
# Получение данных о ценах облигаций
import requests
import csv
from datetime import datetime
import time
def get_bond_data(bond_identifier, start_date, end_date):
"""Получение данных по облигации (ISIN, тикер или название)"""
# Поиск облигации
url = "https://iss.moex.com/iss/securities.json"
params = {'q': bond_identifier}
response = requests.get(url, params=params)
data = response.json()
# Ищем облигацию
bond_info = None
for security in data['securities']['data']:
if (security[1] == bond_identifier or # ISIN
security[0] == bond_identifier or # тикер
security[2] == bond_identifier): # название
bond_info = {
'ticker': security[0],
'shortname': security[1],
'fullname': security[2],
'isin': security[1] if security[1]. <a name="cut"></a> startswith('RU') else security[0],
'boardid': security[4]
}
break
if not bond_info:
print(f"Облигация '{bond_identifier}' не найдена")
return None
print(f"Найдена облигация: {bond_info['fullname']}")
print(f"Тикер: {bond_info['ticker']}, ISIN: {bond_info['isin']}")
# Получаем исторические данные
url = f"https://iss.moex.com/iss/history/engines/stock/markets/bonds/securities/{bond_info['ticker']}.json"
all_data = []
start = 0
page_size = 100
while True:
params = {
'from': start_date.strftime('%Y-%m-%d'),
'till': end_date.strftime('%Y-%m-%d'),
'start': start,
'history.columns': 'TRADEDATE,OPEN,HIGH,LOW,CLOSE,VOLUME,ACCINT'
}
response = requests.get(url, params=params)
data = response.json()
history_data = data['history']['data']
if not history_data:
break
all_data.extend(history_data)
start += page_size
time.sleep(0.1)
return {
'bond_info': bond_info,
'prices': all_data
}
def save_to_csv(prices_data, filename):
"""Сохранение данных в CSV"""
with open(filename, 'w', newline='', encoding='utf-8-sig') as csvfile:
writer = csv.writer(csvfile)
writer.writerow(['Дата', 'Тикер', 'ISIN', 'Название', 'Цена открытия',
'Максимум', 'Минимум', 'Цена закрытия', 'Объем', 'НКД'])
for row in prices_data['prices']:
writer.writerow([
row[0], # TRADEDATE
prices_data['bond_info']['ticker'],
prices_data['bond_info']['isin'],
prices_data['bond_info']['fullname'],
row[1] if row[1] else '',
row[2] if row[2] else '',
row[3] if row[3] else '',
row[4] if row[4] else '',
row[5] if row[5] else '',
row[6] if len(row) > 6 and row[6] else ''
])
def main():
# ВВЕДИТЕ ВАШ ИДЕНТИФИКАТОР ОБЛИГАЦИИ ЗДЕСЬ
bond_identifier = "SU26247RMFS5" # Замените на нужный ISIN, тикер или название
start_date = datetime(2023, 9, 10)
end_date = datetime(2025, 10, 20)
print(f"Поиск облигации: {bond_identifier}")
print(f"Период: {start_date.date()} - {end_date.date()}")
print("-" * 50)
prices_data = get_bond_data(bond_identifier, start_date, end_date)
if prices_data and prices_data['prices']:
print(f"Успешно! Получено {len(prices_data['prices'])} записей")
# Сохраняем в CSV
filename = f'{bond_identifier}_prices.csv'
save_to_csv(prices_data, filename)
print(f"Данные сохранены в {filename}")
# Показываем последние 5 записей
print("\nПоследние 5 торговых дней:")
for row in prices_data['prices'][-5:]:
print(f"{row[0]}: Закрытие {row[4]} руб., Объем {row[5]}, НКД {row[6] if len(row) > 6 else 'N/A'}")
else:
print("Данные не найдены")
if __name__ == "__main__":
main()
Это котировки с задержкой в 15 минут, верно? Пытаюсь найти способ получить поток данных без задержки.