Давайте создадим бота для получения автоматических обновлений на фондовом рынке с помощью Python, Discord и API-интерфейса Lemon.markets.
Уследить за всеми событиями на фондовом рынке может быть сложно. К счастью, как разработчик, мы можем помочь себе с автоматизацией. Однако иногда сложность может сделать это сложной задачей. Это не должно быть так все время. В этом сообщении блога я объясню, как написать скрипт с использованием Python для получения такого автоматического сообщения об акциях, состоящего менее чем из 100 строк.
План
Во-первых, нам нужен план для нашего сценария. Нам нужна программа, которая отправляет нам сообщения о текущих событиях в отношении ETF и акций. Наша программа состоит из 3 разных частей. Первый модуль загружает файл JSON со всеми нашими акциями, которые мы хотим посмотреть. Второй модуль делает запрос к API, а третий и последний модуль создает сообщение из полученных данных. Это сообщение должно регулярно отправляться с таймером или заданием cron. Мы можем использовать бота Discord для получения push-уведомления. Преимущество этого заключается в том, что нам нужно только запросить веб-перехватчик.
Какой API используется?
В этой статье я использую API рыночных данных от немецкого стартапа lemon.markets. Они обеспечивают фантастический опыт разработки и имеют обширную документацию. Тем не менее, это жизнеспособный выбор только в том случае, если вы находитесь в Германии. Для рынка США я бы рекомендовал присмотреться к Альпаке. Но статья более независима от выбора API. Мое внимание сосредоточено на требованиях (ограничение скорости 10 запросов от бесплатного плана в случае лимона.маркеты) и на том, как эффективно использовать данные ресурсы. Итак, приступим:
Часть 1: Загрузка данных
Начнем с первого модуля нашего скрипта: Загрузка файла JSON в диктофон. Наш файл JSON выглядит примерно так:
{
"watchlist": [
{
"isin": "US4581401001",
"name": "Intel"
}
]
}
Эта структура не сложна, и это здорово. Нам не нужно больше. Сначала мы создаем словарь акций. Почему дикт, а не список? Диктатор поможет нам позже, объединяя разные наборы данных. Итак, мы открываем файл, загружаем JSON и получаем массив watchlist. После этого мы перебираем все позиции.
def read_config() -> dict:
stock_list = {}
with open(r"config.json") as file:
data_raw = json.loads(file.read())
watchlist = data_raw["watchlist"]
ISIN или международный идентификационный номер ценных бумаг — это международный идентификатор, присваиваемый каждой акции. Этот идентификатор уникален и идеально подходит для ключа наших словарей. Мы используем ISIN и добавляем имя в качестве значения.
def read_config() -> dict:
stock_list = {}
with open(r"config.json") as file:
data_raw = json.loads(file.read())
watchlist = data_raw["watchlist"]
for stock in watchlist:
stock_list[stock["isin"]] = stock["name"]
return stock_list
Теперь у нас есть диктат с нашими акциями. Перейдем к запросу.
Часть 2. Получите биржевые данные
Мы хотим рассчитать приблизительное изменение цены акций. Для этого нам нужно только знать, как акции вели себя утром и вечером. Здесь мы можем использовать точки данных графика OHLC. Но что такое график OHLC?
Что такое графики OHLC?
График OHLC (Open-High-Low-Close) показывает движение цены финансового инструмента. Он состоит из свечной или гистограммы, которая показывает диапазон цен за определенный период времени. Точки данных для свечи состоят из четырех частей: цена открытия (Open), самая высокая цена (High), самая низкая цена (Low) и цена закрытия (Close).
Нас интересует только цена открытия и цена закрытия дня. Для запроса нам нужен только код ISIN акции и ключ API от Lemon.markets. С помощью d1 мы выбираем диапазон дневного времени. Мы также делаем заполнитель для ISIN, чтобы мы могли заменить их. Кроме того, мы получаем ключ API из переменной среды. Это гарантирует, что мы не упустим случайно ключи API. Наконец, мы конвертируем результат в JSON.
data_day = requests.get(url=f"https://data.lemon.markets/v1/ohlc/d1?isin={isin}",
headers={"Authorization": f"Bearer {os.environ.get('APIKEY')}"}).json()
В идеальном случае мы получаем следующее:
{
'status': 'ok',
'time': '2022-12-26T12:37:50.270+00:00',
'results':
[
{
'isin': 'US00507V1098',
'o': 70.88,
'h': 71.55,
'l': 70.82,
'c': 71.52,
'v': 269,
'pbv': 19210.93,
't': '2022-12-23T00:00:00.000+00:00',
'mic': 'XMUN'
}
],
'previous': None,
'next': None,
'total': 10,
'page': 1,
'pages': 1
}
Нам нужно только перебрать результаты, чтобы получить необходимые данные.
open_price = data_day["results"][0]["o"] close_price = data_day["results"][0]["c"]
Отсюда легко рассчитать прибыли и убытки дня. Мы делим цену открытия на цену закрытия и переводим ее в проценты. Наконец, мы округляем значение до 4 цифр, чтобы в дальнейшем в нашем сообщении не появлялись длинные кривые числа.
gain_loss_price_day = round(((open_price / close_price) * 100) - 100, 4)
В конце нам нужно только составить строку. Кроме того, я включаю последнюю цену акций. Но почему я беру в сообщении юникодные числа вместо реальных символов? Ну, у Discord API могут быть проблемы с этими персонажами, поэтому я беру этот вариант.
Вместе они образуют эту функцию:
def get_message(isin: str, name: str) -> str:
data_day = requests.get(url=f"https://data.lemon.markets/v1/ohlc/d1?isin={isin}",
headers={"Authorization": f"Bearer {os.environ.get('APIKEY')}"}).json()
open_price = data_day["results"][0]["o"]
close_price = data_day["results"][0]["c"]
gain_loss_price_day = round(((open_price / close_price) * 100) - 100, 4)
message = f'{name} -> price: {close_price} \u20ac, 24h: {gain_loss_price_day} \u0025'
return message
Часть 4: Отправка сообщения
Последний модуль, безусловно, самый простой. Нам нужно только сделать POST-запрос Discord. В теле запроса мы включаем наше сообщение. Конечно, вам нужно сначала создать веб-хук на данном сервере. Для этого вы получите специальный URL, который я также поместил в переменную окружения.
def send_discord_message(discord_url: str, discord_message: str):
requests.post(discord_url, json={"content": f'{discord_message}'})
В конце нам нужно только перебрать словарь stock_config и построить сообщения.
Все вместе приводит к этому:
import json
import os
import requests
def read_config() -> dict:
stock_list = {}
with open(r"config.json") as file:
data_raw = json.loads(file.read())
watchlist = data_raw["watchlist"]
for stock in watchlist:
stock_list[stock["isin"]] = stock["name"]
return stock_list
def get_message(isin: str, name: str) -> str:
data_day = requests.get(url=f"https://data.lemon.markets/v1/ohlc/d1?isin={isin}",
headers={"Authorization": f"Bearer {os.environ.get('APIKEY')}"}).json()
open_price = data_day["results"][0]["o"]
close_price = data_day["results"][0]["c"]
gain_loss_price_day = round(((open_price / close_price) * 100) - 100, 4)
message = f'{name} -> price: {close_price} \u20ac, 24h: {gain_loss_price_day} \u0025'
return message
def send_discord_message(discord_url: str, discord_message: str):
requests.post(discord_url, json={"content": f'{discord_message}'})
stock_config = read_config()
message_array = ""
for key, value in stock_config.items():
message_array += get_message(key, value)
send_discord_message(os.environ.get("DISCORDURL"), message_array)
Часть 5: Пакетирование запросов
В общем, теперь у нас есть работающий скрипт. Однако в текущей версии кода мы отправляем запрос на Lemon.market для каждой отдельной акции. Это приведет к тому, что в бесплатном плане будет всего 10 акций в минуту. Но мы можем лучше!
Мы можем запросить до 10 акций за вызов API. Итак, давайте объединим разные запросы. Это означает, что нам нужно реструктурировать второй модуль. Теперь ему нужно построить разные фрагменты из данных конфигурации, обработать данные для каждого фрагмента и объединить все данные вместе. В результате мы заменяем функцию get_message на более специализированные:
def get_data() def merge_data() def single_batch() def build_message()
Создание единого пакетного запроса данных о запасах
Ядро находится в функции get_data. Здесь мы можем позаимствовать большую часть кода из старого get_message. Но как нам группировать запросы? API-интерфейс Lemon.markets позволяет нам одновременно запрашивать несколько кодов ISIN. Нам нужно только связать все ISIN вместе примерно так: isin=US00507V1098,US00724F1012,US0162551016,US02079K3059,US02079K1079,US0231351067,US0079031078,US0255371017,US0311621009,US0326541051.
Эта часть выполняется в самом начале. Мы просто берем ключи из словаря, соединяем их запятой.
def get_data(stock_list_chunk: dict) -> (dict, dict):
isins_list = stock_list_chunk.keys()
isin_string = ','.join(isins_list)
Затем делаем звонок. После этого мы просто перебираем результат запроса и добавляем прибыль и цены закрытия в словарь. Важно, чтобы диктовки имели один и тот же ключ в виде ISIN. В целом, код должен выглядеть знакомым для функции get_message.
def get_data(stock_list_chunk: dict) -> (dict, dict):
isins_list = stock_list_chunk.keys()
isin_string = ','.join(isins_list)
data_day = requests.get(url=f"https://data.lemon.markets/v1/ohlc/d1?isin={isin_string}",
headers={"Authorization": f"Bearer {os.environ.get('APIKEY')}"}).json()
gains_stock = {}
close_prices = {}
for stock in data_day["results"]:
open_price = stock["o"]
close_price = stock["c"]
gain_loss_price_day = round(((open_price / close_price) * 100) - 100, 4)
gains_stock[stock["isin"]] = gain_loss_price_day
close_prices[stock["isin"]] = close_price
return gains_stock, close_prices
Следующим шагом является объединение отдельных словарей. Для этого мы перебираем ключи первого словаря и добавляем оставшиеся элементы в кортеж.
def merge_data(list_chunk: dict, gain_day: dict, end_price_day: dict) -> dict:
merged_dict = {}
for k in list_chunk.keys():
merged_dict[k] = (list_chunk[k], gain_day[k], end_price_day[k])
return merged_dict
В single_batch() объединяем все функции сверху.
def single_batch(chunk: dict) -> dict:
gains, close_prices = get_data(chunk)
merged_data = merge_data(chunk, gains, close_prices)
return merged_data
И последнее, но не менее важное: метод build_message(). Эта функция получает словарь и перебирает его, чтобы создать сообщение большего размера.
def build_message(data_dict: dict) -> str:
messages = ""
for item in data_dict.values():
messages += f'{item[0]} -> price: {item[2]} \u20ac, 24h: {item[1]} \u0025 \n'
return messages
Кроме того, мы добавляем нашу функцию send_discord_message с проверкой ошибок.
def send_discord_message(discord_url: str, discord_message: str):
result = requests.post(discord_url, json={"content": f'{discord_message}'})
try:
result.raise_for_status()
except requests.exceptions.HTTPError as err:
print(err)
else:
print("Payload delivered successfully, code {}.".format(result.status_code))Finally, we also need to batch it properly. So we go over our stockInfo list in the increment of 10 and create a chunk each time. Then this chunk is transferred into a larger data set.
Наконец, нам также нужно правильно его запаковать. Итак, мы просматриваем наш список stock_info с шагом 10 и каждый раз создаем фрагмент. Затем этот фрагмент переносится в больший набор данных.
if __name__ == "__main__":
stock_info = read_config()
all_data = {}
for count in range(0, len(stock_info), 10):
stock_chunk = {k: stock_info[k] for k in list(stock_info)[count:(count + 10)]}
batch = single_batch(stock_chunk)
all_data.update(batch)
message_array = build_message(all_data)
send_discord_message(os.environ.get("DISCORDURL"), message_array)
Теперь код должен выглядеть так:
import json
import os
import requests
# function to read the config file and return a dictionary of ISIN codes and stock names
def read_config() -> dict:
stock_list = {}
with open(r"./config2.json") as file:
data_raw = json.loads(file.read())
watchlist = data_raw["watchlist"]
for stock in watchlist:
stock_list[stock["isin"]] = stock["name"]
return stock_list
# function to get data for a list of ISIN codes
def get_data(stock_list_chunk: dict) -> (dict, dict):
# get the list of ISIN codes and convert the list of ISIN codes to a string separated by commas
isins_list = stock_list_chunk.keys()
isin_string = ','.join(isins_list)
data_day = requests.get(url=f"https://data.lemon.markets/v1/ohlc/d1?isin={isin_string}",
headers={"Authorization": f"Bearer {os.environ.get('APIKEY')}"}).json()
gains_stock = {}
close_prices = {}
# iterate through the results and calculate the gain/loss and the closing price for each stock
for stock in data_day["results"]:
open_price = stock["o"]
close_price = stock["c"]
gain_loss_price_day = round(((open_price / close_price) * 100) - 100, 4)
gains_stock[stock["isin"]] = gain_loss_price_day
close_prices[stock["isin"]] = close_price
return gains_stock, close_prices
# function to merge the stock information, gain/loss, and closing price into a single dictionary
def merge_data(list_chunk: dict, gain_day: dict, end_price_day: dict) -> dict:
merged_dict = {}
# iterate through the list of ISIN codes and create a tuple of the stock information, gain/loss, and closing price
for k in list_chunk.keys():
merged_dict[k] = (list_chunk[k], gain_day[k], end_price_day[k])
return merged_dict
# function to get the data for a single chunk of ISIN codes
def single_batch(chunk: dict) -> dict:
gains, close_prices = get_data(chunk)
merged_data = merge_data(chunk, gains, close_prices)
return merged_data
# function to build a message string from the data dictionary
def build_message(data_dict: dict) -> str:
messages = ""
# iterate through the data and add it to the message string
for item in data_dict.values():
messages += f'{item[0]} -> price: {item[2]} \u20ac, 24h: {item[1]} \u0025 \n'
return messages
# function for sending the message to a discord webhook
def send_discord_message(discord_url: str, discord_message: str):
# post to discrod via webhook
result = requests.post(discord_url, json={"content": f'{discord_message}'})
# check potential errors
try:
result.raise_for_status()
except requests.exceptions.HTTPError as err:
print(err)
else:
print("Payload delivered successfully, code {}.".format(result.status_code))
if __name__ == "__main__":
stock_info = read_config()
all_data = {}
# create diffrent chunk from the stockInfo dict and call the API
for count in range(0, len(stock_info), 10):
stock_chunk = {k: stock_info[k] for k in list(stock_info)[count:(count + 10)]}
batch = single_batch(stock_chunk)
all_data.update(batch)
# build the complete message
message_array = build_message(all_data)
send_discord_message(os.environ.get("DISCORDURL"), message_array)
Автоматизация
Чтобы автоматическое сообщение отправлялось регулярно, необходимо убедиться, что скрипт выполняется ежедневно. Для этого вы можете создать модуль systemd на сервере, который будет ежедневно запускать скрипт. В качестве альтернативы, простое задание Cron также выполняет задачу.
Заключение
Это только начало. Сценарий довольно модульный, поэтому его легко расширить. Отправка сообщения Discord об обновлениях запасов является более или менее заполнителем для некоторых реальных приложений. Это преимущество, когда мыслит более модульно. Каждый модуль можно заменить или переписать, чтобы просматривать что-то другое. Зачем загружать статический JSON. Давайте отследим весь индекс, вызвав другой API. Кроме того, вы можете создать личный обзор победителей и проигравших дня. Или вы можете написать запрос во втором модуле и добавить расчет RSI в третий модуль. Теперь вы можете быть проинформированы, когда какая-либо акция пересекает определенный порог. Или вы создаете небольшой рабочий стол с помощью Rust and Tauri. Возможности безграничны.
Единственное, что я хочу дать на пути, это то, что каким бы большим и сложным ни было приложение, все равно оставляйте четкую и простую структуру. Сделайте код понятным и понятным. Излишняя сложность никогда не решала проблему. Помните, что это область, которая может повлиять на ваш карман.
Вот код этой статьи:
This article can also be found on my blog: https://nikolas.blog/automating/
