330 lines
13 KiB
Python
330 lines
13 KiB
Python
from __future__ import annotations
|
||
|
||
import logging
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, datetime
|
||
from decimal import Decimal
|
||
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import NominalSource
|
||
from app.integrations.gateway import (
|
||
GatewayAuthError,
|
||
GatewayError,
|
||
GatewayNotFoundError,
|
||
)
|
||
from app.integrations.moex import MarketQuote, MoexClient, MoexDataError
|
||
from app.repositories.market_data import MarketDataRepository
|
||
from app.repositories.nominals import NominalHistoryRepository
|
||
from app.repositories.securities import SecurityRepository
|
||
from app.services.request_log import save_request_log
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Источник значения, если действует ручная правка (D6).
|
||
MANUAL_SOURCE = "ручная"
|
||
_REASON_MAX = 500
|
||
|
||
|
||
class MarketDataError(RuntimeError):
|
||
"""Ошибка с понятным пользователю текстом."""
|
||
|
||
|
||
@dataclass
|
||
class MarketDataUpdateResult:
|
||
on_date: date
|
||
updated: int = 0
|
||
no_trades: int = 0
|
||
errors: int = 0
|
||
messages: list[str] = field(default_factory=list)
|
||
|
||
@property
|
||
def ok(self) -> bool:
|
||
return self.errors == 0
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MarketDataRow:
|
||
"""Действующие на дату значения: ручная правка, иначе MOEX (D6)."""
|
||
|
||
isin: str
|
||
name: str
|
||
board: str | None # None — торгов не было, есть только ручная правка
|
||
clean_price: Decimal | None
|
||
clean_price_source: str | None
|
||
yield_value: Decimal | None
|
||
yield_source: str | None
|
||
duration_days: Decimal | None
|
||
duration_source: str | None
|
||
accrued_interest: Decimal | None
|
||
face_value: Decimal | None
|
||
coupon_rate: Decimal | None
|
||
num_trades: int | None
|
||
volume: Decimal | None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ManualQuoteInput:
|
||
isin: str
|
||
market_date: date
|
||
clean_price: Decimal | None
|
||
yield_value: Decimal | None
|
||
duration_days: Decimal | None
|
||
reason: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ManualQuoteRow:
|
||
"""Ручная правка на дату и значения MOEX за ту же дату рядом."""
|
||
|
||
isin: str
|
||
name: str
|
||
clean_price: Decimal | None
|
||
moex_clean_price: Decimal | None
|
||
yield_value: Decimal | None
|
||
moex_yield_value: Decimal | None
|
||
duration_days: Decimal | None
|
||
moex_duration_days: Decimal | None
|
||
reason: str
|
||
updated_by: str
|
||
updated_at: datetime
|
||
|
||
|
||
async def update(
|
||
on_date: date, user: str, *, moex: MoexClient | None = None
|
||
) -> MarketDataUpdateResult:
|
||
"""Котировки на дату по всем активным бумагам.
|
||
|
||
Каждая бумага — своя короткая транзакция: ошибка по одной не откатывает
|
||
и не останавливает остальные. Исключение — ошибки конфигурации шлюза
|
||
(ключ/путь): они одинаковы для всех бумаг, поэтому обновление прерывается.
|
||
Транзакция не держится открытой во время HTTP-запросов.
|
||
Ручные правки (market_data_manual) загрузка не трогает (D6).
|
||
"""
|
||
moex = moex or MoexClient.from_settings()
|
||
result = MarketDataUpdateResult(on_date=on_date)
|
||
|
||
async with SessionLocal() as session:
|
||
securities = list(await SecurityRepository(session).list_active())
|
||
if not securities:
|
||
result.messages.append("Нет активных бумаг для обновления.")
|
||
return result
|
||
|
||
for index, security in enumerate(securities):
|
||
label = f"{security.isin} ({security.name})"
|
||
try:
|
||
quote = moex.get_history(security.secid, on_date)
|
||
if quote is None:
|
||
result.no_trades += 1
|
||
result.messages.append(
|
||
f"{label}: торгов за {on_date:%d.%m.%Y} не было."
|
||
)
|
||
continue
|
||
async with SessionLocal() as session:
|
||
await MarketDataRepository(session).upsert(security.id, quote, user)
|
||
previous_nominal = await _record_moex_nominal(
|
||
NominalHistoryRepository(session), security.id, quote, user
|
||
)
|
||
await session.commit()
|
||
result.updated += 1
|
||
price = (
|
||
f"{quote.clean_price} ({quote.clean_price_source})"
|
||
if quote.clean_price is not None
|
||
else "нет цены"
|
||
)
|
||
message = f"{label}: {quote.board}, цена {price}."
|
||
if previous_nominal is not None:
|
||
message += (
|
||
f" Номинал изменился: {_plain(previous_nominal)} -> "
|
||
f"{_plain(quote.face_value)}."
|
||
)
|
||
result.messages.append(message)
|
||
except (GatewayAuthError, GatewayNotFoundError) as exc:
|
||
remaining = len(securities) - index
|
||
result.errors += remaining
|
||
result.messages.append(f"{label}: {exc}")
|
||
if remaining > 1:
|
||
result.messages.append(
|
||
f"Обновление прервано: остальные {remaining - 1} бумаг "
|
||
"не обновлялись — сначала исправьте настройки шлюза."
|
||
)
|
||
break
|
||
except (GatewayError, MoexDataError) as exc:
|
||
result.errors += 1
|
||
result.messages.append(f"{label}: {exc}")
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.exception("Ошибка обновления котировки %s", security.isin)
|
||
result.errors += 1
|
||
result.messages.append(f"{label}: непредвиденная ошибка ({exc}).")
|
||
finally:
|
||
await save_request_log(moex.gateway, user)
|
||
|
||
return result
|
||
|
||
|
||
async def _record_moex_nominal(
|
||
nominals: NominalHistoryRepository,
|
||
security_id: int,
|
||
quote: MarketQuote,
|
||
user: str,
|
||
) -> Decimal | None:
|
||
"""FACEVALUE -> история номинала (D11), если он отличается от последнего
|
||
значения MOEX на эту дату. Ручные записи в сравнении не участвуют: ручной
|
||
номинал действует до следующего изменения номинала на MOEX.
|
||
|
||
Возвращает прежний номинал MOEX, если он был и изменился.
|
||
"""
|
||
if quote.face_value is None:
|
||
return None
|
||
last = await nominals.last_moex_on(security_id, quote.trade_date)
|
||
if last is not None and last.face_value == quote.face_value:
|
||
return None
|
||
await nominals.upsert(
|
||
security_id=security_id,
|
||
valid_from=quote.trade_date,
|
||
face_value=quote.face_value,
|
||
source=NominalSource.MOEX,
|
||
reason=None,
|
||
user=user,
|
||
)
|
||
return last.face_value if last is not None else None
|
||
|
||
|
||
def _plain(value: Decimal | None) -> str:
|
||
return format(value.normalize(), "f") if value is not None else "—"
|
||
|
||
|
||
def _pick(
|
||
manual: Decimal | None, moex: Decimal | None, moex_source: str | None
|
||
) -> tuple[Decimal | None, str | None]:
|
||
if manual is not None:
|
||
return manual, MANUAL_SOURCE
|
||
if moex is not None:
|
||
return moex, moex_source
|
||
return None, None
|
||
|
||
|
||
async def list_for_date(market_date: date) -> list[MarketDataRow]:
|
||
"""Котировки на дату с учётом ручных правок (D6)."""
|
||
async with SessionLocal() as session:
|
||
repository = MarketDataRepository(session)
|
||
quotes = await repository.list_for_date(market_date)
|
||
overrides = await repository.list_manual_for_date(market_date)
|
||
|
||
securities = {security.id: security for _, security in [*quotes, *overrides]}
|
||
moex = {security.id: row for row, security in quotes}
|
||
manual = {security.id: row for row, security in overrides}
|
||
rows = []
|
||
for security_id, security in securities.items():
|
||
quote = moex.get(security_id)
|
||
override = manual.get(security_id)
|
||
price, price_source = _pick(
|
||
override.clean_price if override else None,
|
||
quote.clean_price if quote else None,
|
||
quote.clean_price_source if quote else None,
|
||
)
|
||
yield_value, yield_source = _pick(
|
||
override.yield_value if override else None,
|
||
quote.yield_value if quote else None,
|
||
quote.yield_source if quote else None,
|
||
)
|
||
duration, duration_source = _pick(
|
||
override.duration_days if override else None,
|
||
quote.duration_days if quote else None,
|
||
"DURATION",
|
||
)
|
||
rows.append(
|
||
MarketDataRow(
|
||
isin=security.isin,
|
||
name=security.name,
|
||
board=quote.board if quote else None,
|
||
clean_price=price,
|
||
clean_price_source=price_source,
|
||
yield_value=yield_value,
|
||
yield_source=yield_source,
|
||
duration_days=duration,
|
||
duration_source=duration_source,
|
||
accrued_interest=quote.accrued_interest if quote else None,
|
||
face_value=quote.face_value if quote else None,
|
||
coupon_rate=quote.coupon_rate if quote else None,
|
||
num_trades=quote.num_trades if quote else None,
|
||
volume=quote.volume if quote else None,
|
||
)
|
||
)
|
||
return sorted(rows, key=lambda row: row.isin)
|
||
|
||
|
||
def validate_manual(value: ManualQuoteInput) -> list[str]:
|
||
errors = []
|
||
if (
|
||
value.clean_price is None
|
||
and value.yield_value is None
|
||
and value.duration_days is None
|
||
):
|
||
errors.append("Укажите цену, доходность или дюрацию.")
|
||
if value.clean_price is not None and value.clean_price <= 0:
|
||
errors.append("Цена должна быть больше нуля.")
|
||
if value.duration_days is not None and value.duration_days < 0:
|
||
errors.append("Дюрация не может быть отрицательной.")
|
||
reason = value.reason.strip()
|
||
if not reason:
|
||
errors.append("Укажите причину правки.")
|
||
elif len(reason) > _REASON_MAX:
|
||
errors.append(f"Причина — не длиннее {_REASON_MAX} символов.")
|
||
if value.market_date > date.today():
|
||
errors.append("Дата не может быть в будущем.")
|
||
return errors
|
||
|
||
|
||
async def set_manual_quote(value: ManualQuoteInput, user: str) -> None:
|
||
"""Ручные значения на дату (D6). Загрузка MOEX их не перезаписывает.
|
||
|
||
TODO(BIQ-38115): права на ручные значения — Q16, пока может любой
|
||
пользователь; цена при отсутствии торгов — Q14.
|
||
"""
|
||
errors = validate_manual(value)
|
||
if errors:
|
||
raise MarketDataError(" ".join(errors))
|
||
async with SessionLocal() as session:
|
||
security = await SecurityRepository(session).get_by_isin(value.isin)
|
||
if security is None:
|
||
raise MarketDataError(f"Бумаги {value.isin} нет в справочнике.")
|
||
await MarketDataRepository(session).upsert_manual(
|
||
security_id=security.id,
|
||
market_date=value.market_date,
|
||
clean_price=value.clean_price,
|
||
yield_value=value.yield_value,
|
||
duration_days=value.duration_days,
|
||
reason=value.reason.strip(),
|
||
user=user,
|
||
)
|
||
await session.commit()
|
||
|
||
|
||
async def list_manual_for_date(market_date: date) -> list[ManualQuoteRow]:
|
||
async with SessionLocal() as session:
|
||
repository = MarketDataRepository(session)
|
||
overrides = await repository.list_manual_for_date(market_date)
|
||
quotes = {
|
||
security.id: row
|
||
for row, security in await repository.list_for_date(market_date)
|
||
}
|
||
rows = []
|
||
for override, security in overrides:
|
||
quote = quotes.get(security.id)
|
||
rows.append(
|
||
ManualQuoteRow(
|
||
isin=security.isin,
|
||
name=security.name,
|
||
clean_price=override.clean_price,
|
||
moex_clean_price=quote.clean_price if quote else None,
|
||
yield_value=override.yield_value,
|
||
moex_yield_value=quote.yield_value if quote else None,
|
||
duration_days=override.duration_days,
|
||
moex_duration_days=quote.duration_days if quote else None,
|
||
reason=override.reason,
|
||
updated_by=override.updated_by,
|
||
updated_at=override.updated_at,
|
||
)
|
||
)
|
||
return rows
|