54 lines
1.4 KiB
Python
54 lines
1.4 KiB
Python
# app/src/integrations/exchange/market_cache.py
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from src.core.config import load_settings
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class MarketPriceSnapshot:
|
|
symbol: str
|
|
price: float
|
|
bid_price: float | None
|
|
ask_price: float | None
|
|
updated_at: str
|
|
source: str = "market-cache"
|
|
|
|
|
|
class MarketPriceCache:
|
|
_prices: dict[str, MarketPriceSnapshot] = {}
|
|
|
|
# сохранить последнюю цену
|
|
@classmethod
|
|
def set_price(
|
|
cls,
|
|
*,
|
|
symbol: str,
|
|
price: float,
|
|
bid_price: float | None = None,
|
|
ask_price: float | None = None,
|
|
updated_at: str | None = None,
|
|
source: str = "market-polling",
|
|
) -> None:
|
|
settings = load_settings()
|
|
|
|
if updated_at is None:
|
|
updated_at = datetime.now(ZoneInfo(settings.tz)).strftime("%d.%m.%Y %H:%M:%S")
|
|
|
|
cls._prices[symbol.upper()] = MarketPriceSnapshot(
|
|
symbol=symbol.upper(),
|
|
price=price,
|
|
bid_price=bid_price,
|
|
ask_price=ask_price,
|
|
updated_at=updated_at,
|
|
source=source,
|
|
)
|
|
|
|
# получить последнюю цену
|
|
@classmethod
|
|
def get_price(cls, symbol: str) -> MarketPriceSnapshot | None:
|
|
return cls._prices.get(symbol.upper()) |