LAB5/scripts/fetch_rates.py - master - antonnovom/hw_bit

LAB5/scripts/fetch_rates.py
84 строки3 KB
antonnovom
upload files
06 май 2025, 18:08
06 май 2025, 18:0807681f8
import os
import time
import uuid
import requests
from datetime import datetime, timezone
from db_utils import insert_rates
from dotenv import load_dotenv

load_dotenv()

API_URL_NOW      = "https://api.coingecko.com/api/v3/simple/price"
API_URL_HISTORY  = "https://api.coingecko.com/api/v3/coins/{id}/market_chart"

# Список монет (можно менять в .env, если ранее сделали CRYPTO_IDS)
CRYPTO_IDS       = os.getenv('CRYPTO_IDS', 'bitcoin,ethereum,litecoin').split(',')
VS_CURRENCY      = os.getenv('VS_CURRENCY', 'usd')

# Интервал текущих цен в секундах
INTERVAL         = int(os.getenv('FETCH_INTERVAL', 3600))

# Исторический режим
HISTORICAL       = os.getenv('HISTORICAL', 'false').lower() == 'true'
HISTORICAL_DAYS  = int(os.getenv('HISTORICAL_DAYS', 7))


def fetch_current_prices(ids, vs):
    """Текущие цены (Simple Price API)."""
    params = {'ids': ','.join(ids), 'vs_currencies': vs}
    resp = requests.get(API_URL_NOW, params=params, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    ts = datetime.now(timezone.utc)
    records = []
    for symbol, info in data.items():
        price = info.get(vs)
        if price is not None:
            records.append((str(uuid.uuid4()), symbol, float(price), ts))
    return records


def fetch_historical_prices(ids, vs, days):
    """
    Исторические цены за последние `days` дней.
    Возвращает список записей (uuid, symbol, price_usd, timestamp).
    """
    records = []
    for coin in ids:
        url = API_URL_HISTORY.format(id=coin)
        params = {'vs_currency': vs, 'days': days}
        resp = requests.get(url, params=params, timeout=10)
        resp.raise_for_status()
        data = resp.json().get('prices', [])
        for timestamp_ms, price in data:
            ts = datetime.fromtimestamp(timestamp_ms / 1000.0, timezone.utc)
            records.append((str(uuid.uuid4()), coin, float(price), ts))
        print(f"  загружено {len(data)} точек для {coin}")
    return records


def main():
    print("=== Сбор курсов криптовалют ===")
    print("Монеты:", CRYPTO_IDS)
    print("Текущий интервал (с):", INTERVAL)
    if HISTORICAL:
        print(f"=== Fetch historical: last {HISTORICAL_DAYS} days ===")
        try:
            hist = fetch_historical_prices(CRYPTO_IDS, VS_CURRENCY, HISTORICAL_DAYS)
            insert_rates(hist)
            print(f"[{datetime.now(timezone.utc)}] Вставлено исторических записей: {len(hist)}")
        except Exception as e:
            print("Ошибка исторического загрузчика:", e)
    print("=== Начало бесконечного сбора текущих цен ===")
    while True:
        try:
            recs = fetch_current_prices(CRYPTO_IDS, VS_CURRENCY)
            insert_rates(recs)
            print(f"[{datetime.now(timezone.utc)}] Вставлено {len(recs)} текущих записей")
        except Exception as err:
            print("Сбой при сборе/вставке:", err)
        time.sleep(INTERVAL)


if __name__ == "__main__":
    main()