Автоматизация процессов обмена валют позволяет создавать надежные и удобные решения для финансовых приложений, которые требуют актуальных данных о курсах валют. Скрипт обменника валют с поддержкой реального времени должен автоматически обновлять курсы, обрабатывать пользовательские запросы и обеспечивать стабильную работу даже при высокой нагрузке. В этой статье мы разберем, как создать автоматизированный скрипт на Python с использованием API для получения курсов валют в реальном времени, с акцентом на автоматизацию и масштабируемость.
Шаг 1: Определение требований
Для создания скрипта с поддержкой реального времени нам нужно:
-
Получать актуальные курсы валют через API.
-
Автоматически обновлять курсы с заданной периодичностью.
-
Предоставлять пользователю возможность конвертировать валюты через консоль или веб-интерфейс.
-
Обеспечить обработку ошибок и логирование для мониторинга.
Мы будем использовать ExchangeRate-API для получения данных и библиотеку schedule для планирования обновлений курсов. Также мы добавим веб-интерфейс с использованием FastAPI для удобного взаимодействия.
Шаг 2: Настройка окружения
Установка зависимостей
Установим необходимые библиотеки:
pip install requests python-dotenv schedule fastapi uvicorn cachetools-
requests: Для запросов к API.
-
python-dotenv: Для безопасного хранения API-ключа.
-
schedule: Для периодического обновления курсов.
-
fastapi и uvicorn: Для создания веб-интерфейса.
-
cachetools: Для кэширования данных.
Безопасное хранение API-ключа
Создайте файл .env в корне проекта:
API_KEY=ваш_ключ_здесьДобавьте .env в .gitignore, чтобы исключить его из системы контроля версий.
Шаг 3: Создание базового скрипта
Мы создадим скрипт, который:
-
Периодически обновляет курсы валют.
-
Кэширует данные для минимизации запросов к API.
-
Предоставляет API-эндпоинт для конвертации валют.
Создайте файл realtime_currency_exchange.py:
import requests
import schedule
import time
import threading
from cachetools import TTLCache
from dotenv import load_dotenv
import os
from fastapi import FastAPI, HTTPException
import uvicorn
import logging
# Настройка логирования
logging.basicConfig(filename="exchange.log", level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s")
# Кэш на 1 час
cache = TTLCache(maxsize=1, ttl=3600)
# Загрузка API-ключа
def load_api_key():
load_dotenv()
api_key = os.getenv("API_KEY")
if not api_key:
raise ValueError("API ключ не найден в файле .env")
return api_key
# Получение курсов валют
def get_exchange_rates(api_key):
try:
url = f"https://v6.exchangerate-api.com/v6/{api_key}/latest/USD"
response = requests.get(url, timeout=5)
response.raise_for_status()
data = response.json()
if data["result"] == "success":
cache["rates"] = data["conversion_rates"]
logging.info("Курсы валют успешно обновлены")
else:
logging.error("Ошибка API: " + data.get("error-type", "Неизвестная ошибка"))
except requests.exceptions.RequestException as e:
logging.error(f"Ошибка сети: {e}")
# Периодическое обновление курсов
def schedule_rate_updates(api_key):
schedule.every(1).hours.do(get_exchange_rates, api_key=api_key)
while True:
schedule.run_pending()
time.sleep(60)
# Конвертация валют
def convert_currency(amount, from_currency, to_currency):
rates = cache.get("rates")
if not rates:
raise ValueError("Курсы валют недоступны")
try:
if not isinstance(amount, (int, float)) or amount <= 0:
raise ValueError("Сумма должна быть положительным числом")
if from_currency not in rates or to_currency not in rates:
raise ValueError("Неверный код валюты")
if from_currency != "USD":
amount = amount / rates[from_currency]
converted_amount = amount * rates[to_currency]
return round(converted_amount, 2)
except ZeroDivisionError:
raise ValueError("Курс валюты равен нулю")
# FastAPI приложение
app = FastAPI(title="Currency Exchange API")
@app.get("/rates")
async def get_rates():
rates = cache.get("rates")
if not rates:
raise HTTPException(status_code=503, detail="Курсы валют недоступны")
return {"currencies": list(rates.keys()), "rates": rates}
@app.post("/convert")
async def convert(data: dict):
try:
amount = float(data["amount"])
from_currency = data["from_currency"].upper()
to_currency = data["to_currency"].upper()
result = convert_currency(amount, from_currency, to_currency)
logging.info(f"Конвертация: {amount} {from_currency} -> {result} {to_currency}")
return {"result": result, "to_currency": to_currency}
except ValueError as ve:
logging.error(f"Ошибка ввода: {ve}")
raise HTTPException(status_code=400, detail=str(ve))
except Exception as e:
logging.error(f"Произошла ошибка: {e}")
raise HTTPException(status_code=500, detail="Внутренняя ошибка сервера")
# Запуск обновления курсов в фоновом потоке
def start_background_updates(api_key):
thread = threading.Thread(target=schedule_rate_updates, args=(api_key,), daemon=True)
thread.start()
if __name__ == "__main__":
api_key = load_api_key()
get_exchange_rates(api_key) # Первичное получение курсов
start_background_updates(api_key) # Запуск обновлений
uvicorn.run(app, host="0.0.0.0", port=8000)Объяснение кода
-
Автоматизация обновления курсов:
-
Библиотека schedule обновляет курсы каждый час с помощью функции schedule_rate_updates.
-
Обновления выполняются в фоновом потоке, чтобы не блокировать основной процесс.
-
-
Кэширование:
-
cachetools.TTLCache хранит курсы валют в памяти, минимизируя запросы к API.
-
-
Веб-интерфейс:
-
FastAPI предоставляет два эндпоинта:
-
/rates: Возвращает список доступных валют и их курсы.
-
/convert: Выполняет конвертацию на основе JSON-запроса.
-
-
-
Логирование:
-
Все операции и ошибки записываются в файл exchange.log для мониторинга.
-
-
Обработка ошибок:
-
Проверяется корректность ввода, доступность курсов и сетевые сбои.
-
Шаг 4: Использование скрипта
-
Запуск скрипта:
python realtime_currency_exchange.py -
Доступ к API:
-
Откройте браузер или используйте curl:
-
Получение курсов: curl http://localhost:8000/rates
-
Конвертация: curl -X POST -H «Content-Type: application/json» -d ‘{«amount»: 100, «from_currency»: «EUR», «to_currency»: «RUB»}’ http://localhost:8000/convert
-
-
-
Пример ответа:
{"result": 8500.0, "to_currency": "RUB"}
Шаг 5: Оптимизация и надежность
5.1. Ограничение скорости запросов
Чтобы не превышать лимиты API, добавим ratelimiter:
Установка:
pip install ratelimiterОбновление функции:
from ratelimiter import RateLimiter
rate_limiter = RateLimiter(max_calls=10, period=60)
def get_exchange_rates(api_key):
with rate_limiter:
if "rates" in cache:
return
try:
url = f"https://v6.exchangerate-api.com/v6/{api_key}/latest/USD"
response = requests.get(url, timeout=5)
response.raise_for_status()
data = response.json()
if data["result"] == "success":
cache["rates"] = data["conversion_rates"]
logging.info("Курсы валют успешно обновлены")
except requests.exceptions.RequestException as e:
logging.error(f"Ошибка сети: {e}")5.2. Асинхронные запросы
Для повышения производительности можно заменить requests на aiohttp для асинхронных запросов:
import aiohttp
async def get_exchange_rates_async(api_key):
with rate_limiter:
if "rates" in cache:
return
try:
async with aiohttp.ClientSession() as session:
async with session.get(f"https://v6.exchangerate-api.com/v6/{api_key}/latest/USD", timeout=5) as response:
response.raise_for_status()
data = await response.json()
if data["result"] == "success":
cache["rates"] = data["conversion_rates"]
logging.info("Курсы валют успешно обновлены")
except aiohttp.ClientError as e:
logging.error(f"Ошибка сети: {e}")Интегрируйте это в schedule_rate_updates с помощью asyncio.
Шаг 6: Тестирование
-
Тест обновления курсов:
-
Проверьте файл exchange.log, чтобы убедиться, что курсы обновляются каждый час.
-
-
Тест API:
-
Отправьте несколько запросов к /convert с разными валютами и суммами.
-
Попробуйте некорректные данные (например, отрицательные суммы или несуществующие валюты).
-
-
Тест надежности:
-
Отключите интернет и проверьте, как скрипт обрабатывает сбои.
-
Убедитесь, что кэш позволяет выполнять конвертации без новых запросов.
-
Шаг 7: Дополнительные улучшения
-
Фронтенд:
-
Добавьте HTML-страницу с JavaScript для взаимодействия с API.
-
-
Мониторинг:
-
Настройте уведомления (например, через email) при сбоях API.
-
-
Кэширование на диске:
-
Сохраняйте курсы в файл с помощью pickle, чтобы использовать их при перезапуске.
-
Пример кэширования на диске:
import pickle
def save_rates_to_file():
rates = cache.get("rates")
if rates:
with open("rates_cache.pkl", "wb") as f:
pickle.dump(rates, f)
def load_rates_from_file():
try:
with open("rates_cache.pkl", "rb") as f:
cache["rates"] = pickle.load(f)
except FileNotFoundError:
passЗаключение
скрипт обменника валют с поддержкой реального времени, представленный в этой статье, демонстрирует, как автоматизировать получение и обновление курсов валют, обеспечивая высокую производительность и надежность. Использование кэширования, асинхронных запросов и веб-интерфейса делает его подходящим для реальных приложений. Вы можете расширить функциональность, добавив фронтенд, мониторинг или поддержку нескольких API, чтобы адаптировать скрипт под ваши задачи. Автоматизация позволяет минимизировать ручной труд и обеспечить актуальность данных в реальном времени.
