509 lines
18 KiB
Python
509 lines
18 KiB
Python
"""Загрузка выгрузки QUIK в портфель (D17, D19, D20).
|
||
|
||
Предпросмотр — какие строки пойдут в выбранный портфель, какие ждут решения,
|
||
повтор файла, сопоставление с бумагами, итоги — и сохранение: import_batch и
|
||
операции одной транзакцией. Каждая строка — отдельная операция (лотов нет,
|
||
D20); дедупликации по строкам нет — только защита от повтора файла (D19).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections import Counter, defaultdict
|
||
from collections.abc import Iterable, Mapping, Sequence
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, datetime
|
||
from decimal import Decimal
|
||
|
||
from sqlalchemy.exc import IntegrityError
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import ImportBatch
|
||
from app.imports.quik import SOURCE, QuikParseResult, QuikTrade
|
||
from app.repositories.import_batches import ImportBatchRepository
|
||
from app.repositories.operations import OperationRepository
|
||
from app.repositories.reference import ReferenceRepository
|
||
from app.repositories.securities import SecurityRepository
|
||
|
||
# Комментарий сделки -> портфель. /trde сюда сознательно не входит (D19):
|
||
# такие строки — «без портфеля», решает пользователь.
|
||
COMMENT_PORTFOLIO = {"/uw": "UW", "/pp": "SALES"}
|
||
|
||
INCLUDE = "include"
|
||
EXCLUDE = "exclude"
|
||
|
||
|
||
class QuikImportError(RuntimeError):
|
||
"""Отказ в сохранении с понятным пользователю текстом."""
|
||
|
||
|
||
class RowStatus:
|
||
OWN = "own" # комментарий выбранного портфеля
|
||
OTHER = "other" # комментарий другого портфеля — не берём
|
||
UNKNOWN_PENDING = "unknown_pending" # без портфеля, решения ещё нет
|
||
UNKNOWN_INCLUDED = "unknown_included" # без портфеля, отнесена к выбранному
|
||
UNKNOWN_EXCLUDED = "unknown_excluded" # без портфеля, явно исключена
|
||
LOADED_ELSEWHERE = "loaded_elsewhere" # из этого файла уже загружена в другой
|
||
LOADED_HERE = "loaded_here" # из этого файла уже загружена в выбранный
|
||
|
||
|
||
TO_LOAD = frozenset({RowStatus.OWN, RowStatus.UNKNOWN_INCLUDED})
|
||
UNKNOWN = frozenset(
|
||
{
|
||
RowStatus.UNKNOWN_PENDING,
|
||
RowStatus.UNKNOWN_INCLUDED,
|
||
RowStatus.UNKNOWN_EXCLUDED,
|
||
}
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PortfolioRef:
|
||
id: int
|
||
code: str
|
||
name: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class BatchInfo:
|
||
portfolio_code: str
|
||
file_name: str
|
||
date_from: date
|
||
date_to: date
|
||
rows_loaded: int
|
||
rows_excluded: int
|
||
loaded_rows: frozenset[int]
|
||
created_at: datetime
|
||
created_by: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SecurityMatch:
|
||
security_id: int
|
||
isin: str
|
||
|
||
|
||
@dataclass
|
||
class ImportContext:
|
||
"""Всё, что экрану нужно из БД, — одним запросом на перерисовку."""
|
||
|
||
portfolios: list[PortfolioRef]
|
||
same_file: list[BatchInfo] = field(default_factory=list)
|
||
overlapping: list[BatchInfo] = field(default_factory=list)
|
||
securities: dict[str, list[SecurityMatch]] = field(default_factory=dict)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PlannedRow:
|
||
trade: QuikTrade
|
||
status: str
|
||
hint: str | None # портфель по комментарию
|
||
loaded_in: str | None # куда уже загружена из этого файла
|
||
securities: tuple[SecurityMatch, ...]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class FileSummary:
|
||
rows_total: int
|
||
loaded: dict[str, int] # код портфеля -> загружено строк
|
||
nowhere: int
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TotalRow:
|
||
operation: str
|
||
currency: str
|
||
trades: int
|
||
quantity: Decimal
|
||
amount: Decimal
|
||
settlement_amount: Decimal
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CommissionTotal:
|
||
currency: str
|
||
commission_ts: Decimal
|
||
commission_clearing: Decimal
|
||
commission_exchange: Decimal
|
||
commission_tc: Decimal
|
||
commission_broker: Decimal
|
||
|
||
|
||
@dataclass
|
||
class ImportPlan:
|
||
portfolio_code: str
|
||
rows: list[PlannedRow]
|
||
duplicate: BatchInfo | None
|
||
overlapping: list[BatchInfo]
|
||
summary: FileSummary
|
||
|
||
def with_status(self, *statuses: str) -> list[PlannedRow]:
|
||
return [row for row in self.rows if row.status in statuses]
|
||
|
||
@property
|
||
def to_load(self) -> list[PlannedRow]:
|
||
return [row for row in self.rows if row.status in TO_LOAD]
|
||
|
||
@property
|
||
def pending(self) -> int:
|
||
return len(self.with_status(RowStatus.UNKNOWN_PENDING))
|
||
|
||
@property
|
||
def excluded(self) -> int:
|
||
return len(self.with_status(RowStatus.UNKNOWN_EXCLUDED))
|
||
|
||
@property
|
||
def unmatched_instruments(self) -> dict[str, int]:
|
||
"""Краткие имена без бумаги в справочнике -> сколько строк (среди
|
||
тех, что пойдут в портфель или ждут решения)."""
|
||
counter = Counter(
|
||
row.trade.instrument
|
||
for row in self.rows
|
||
if not row.securities
|
||
and (row.status in TO_LOAD or row.status == RowStatus.UNKNOWN_PENDING)
|
||
)
|
||
return dict(sorted(counter.items()))
|
||
|
||
@property
|
||
def ambiguous_instruments(self) -> dict[str, list[str]]:
|
||
"""Одно краткое имя — несколько бумаг. ISIN = выпуск (D20), так быть
|
||
не должно — сохранение запрещено, пока справочник не исправят."""
|
||
return {
|
||
row.trade.instrument: [s.isin for s in row.securities]
|
||
for row in self.rows
|
||
if len(row.securities) > 1
|
||
}
|
||
|
||
@property
|
||
def blockers(self) -> list[str]:
|
||
"""Почему сохранять нельзя; пусто — можно."""
|
||
code = self.portfolio_code
|
||
reasons = []
|
||
if self.duplicate is not None:
|
||
reasons.append(f"этот файл уже загружен в {code}")
|
||
if self.pending:
|
||
reasons.append(f"не решено строк без портфеля: {self.pending}")
|
||
if not self.to_load and self.duplicate is None:
|
||
reasons.append(f"нет строк для загрузки в {code}")
|
||
if self.unmatched_instruments:
|
||
reasons.append(
|
||
"нет в справочнике бумаг: " + ", ".join(self.unmatched_instruments)
|
||
)
|
||
if self.ambiguous_instruments:
|
||
reasons.append(
|
||
"одному краткому имени соответствует несколько бумаг: "
|
||
+ ", ".join(self.ambiguous_instruments)
|
||
)
|
||
return reasons
|
||
|
||
@property
|
||
def ready(self) -> bool:
|
||
"""Можно сохранять: не повтор, всё решено, бумаги найдены однозначно."""
|
||
return not self.blockers
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SaveResult:
|
||
batch_id: int
|
||
portfolio_code: str
|
||
rows_loaded: int
|
||
rows_excluded: int
|
||
|
||
|
||
def portfolio_hint(comment: str) -> str | None:
|
||
return COMMENT_PORTFOLIO.get(comment.strip().casefold())
|
||
|
||
|
||
def suggest_portfolio(trades: Iterable[QuikTrade]) -> str | None:
|
||
"""Портфель по самому частому комментарию; при равенстве — None."""
|
||
counts = Counter(
|
||
hint for hint in (portfolio_hint(t.comment) for t in trades) if hint
|
||
).most_common()
|
||
if not counts or (len(counts) > 1 and counts[0][1] == counts[1][1]):
|
||
return None
|
||
return counts[0][0]
|
||
|
||
|
||
async def load_context(parsed: QuikParseResult) -> ImportContext:
|
||
async with SessionLocal() as session:
|
||
return await _read_context(session, parsed)
|
||
|
||
|
||
async def _read_context(
|
||
session: AsyncSession, parsed: QuikParseResult
|
||
) -> ImportContext:
|
||
content_hash = parsed.content_hash
|
||
portfolios = [
|
||
PortfolioRef(p.id, p.code, p.name)
|
||
for p in await ReferenceRepository(session).list_portfolios()
|
||
]
|
||
context = ImportContext(portfolios=portfolios)
|
||
if content_hash is None:
|
||
return context
|
||
|
||
batches = ImportBatchRepository(session)
|
||
context.same_file = [
|
||
_batch_info(batch, code)
|
||
for batch, code in await batches.list_by_content(content_hash)
|
||
]
|
||
context.overlapping = [
|
||
_batch_info(batch, code)
|
||
for batch, code in await batches.list_overlapping(
|
||
parsed.date_from, parsed.date_to, content_hash
|
||
)
|
||
]
|
||
|
||
names = sorted({t.instrument for t in parsed.trades})
|
||
by_name: dict[str, list[SecurityMatch]] = defaultdict(list)
|
||
for security in await SecurityRepository(session).list_by_names(names):
|
||
by_name[security.name].append(SecurityMatch(security.id, security.isin))
|
||
context.securities = dict(by_name)
|
||
return context
|
||
|
||
|
||
async def save_import(
|
||
parsed: QuikParseResult,
|
||
*,
|
||
file_name: str,
|
||
portfolio_code: str,
|
||
decisions: Mapping[int, str],
|
||
expected_rows: Sequence[int],
|
||
user: str,
|
||
) -> SaveResult:
|
||
"""import_batch + операции одной транзакцией.
|
||
|
||
План строится заново по свежим данным БД. expected_rows — порядковые
|
||
номера строк, которые пользователь видел в предпросмотре: если набор
|
||
изменился (например, файл тем временем загрузили в другой портфель),
|
||
ничего не пишем — пусть проверит ещё раз.
|
||
"""
|
||
if parsed.content_hash is None or parsed.date_from is None:
|
||
raise QuikImportError("Файл с ошибками не загружается.")
|
||
|
||
async with SessionLocal() as session:
|
||
context = await _read_context(session, parsed)
|
||
plan = build_plan(parsed.trades, context, portfolio_code, decisions)
|
||
if plan.blockers:
|
||
raise QuikImportError("Загрузить нельзя: " + "; ".join(plan.blockers) + ".")
|
||
rows = plan.to_load
|
||
if [row.trade.ordinal for row in rows] != list(expected_rows):
|
||
raise QuikImportError(
|
||
"Данные по файлу изменились, пока вы его проверяли (например, "
|
||
"его загрузили в другой портфель). Проверьте предпросмотр ещё раз."
|
||
)
|
||
|
||
portfolio_id = next(
|
||
p.id for p in context.portfolios if p.code == portfolio_code
|
||
)
|
||
reference = ReferenceRepository(session)
|
||
type_ids = await reference.operation_type_ids()
|
||
currency_ids = await reference.currency_ids()
|
||
missing = sorted({row.trade.currency for row in rows} - set(currency_ids))
|
||
if missing:
|
||
raise QuikImportError(
|
||
"Валюты нет в справочнике: " + ", ".join(missing) + "."
|
||
)
|
||
|
||
try:
|
||
batch = await ImportBatchRepository(session).add(
|
||
source=SOURCE,
|
||
portfolio_id=portfolio_id,
|
||
file_name=file_name,
|
||
content_sha256=parsed.content_hash,
|
||
date_from=parsed.date_from,
|
||
date_to=parsed.date_to,
|
||
rows_total=len(parsed.trades),
|
||
loaded_rows=[row.trade.ordinal for row in rows],
|
||
rows_excluded=plan.excluded,
|
||
user=user,
|
||
)
|
||
await OperationRepository(session).add_many(
|
||
[
|
||
_operation_values(
|
||
row,
|
||
portfolio_id=portfolio_id,
|
||
operation_type_id=type_ids[row.trade.operation],
|
||
currency_id=currency_ids[row.trade.currency],
|
||
batch_id=batch.id,
|
||
user=user,
|
||
)
|
||
for row in rows
|
||
]
|
||
)
|
||
await session.commit()
|
||
except IntegrityError as exc:
|
||
await session.rollback()
|
||
raise QuikImportError(
|
||
f"Этот файл уже загружен в {portfolio_code}."
|
||
) from exc
|
||
return SaveResult(
|
||
batch_id=batch.id,
|
||
portfolio_code=portfolio_code,
|
||
rows_loaded=len(rows),
|
||
rows_excluded=plan.excluded,
|
||
)
|
||
|
||
|
||
def _operation_values(
|
||
row: PlannedRow,
|
||
*,
|
||
portfolio_id: int,
|
||
operation_type_id: int,
|
||
currency_id: int,
|
||
batch_id: int,
|
||
user: str,
|
||
) -> dict[str, object]:
|
||
trade = row.trade
|
||
[security] = row.securities # однозначность проверена в blockers
|
||
return {
|
||
"security_id": security.security_id,
|
||
"portfolio_id": portfolio_id,
|
||
"operation_type_id": operation_type_id,
|
||
"currency_id": currency_id,
|
||
"trade_date": trade.trade_date,
|
||
"trade_time": trade.trade_time,
|
||
"settlement_date": trade.settlement_date,
|
||
"account": trade.account or None,
|
||
"quantity": trade.quantity,
|
||
"price": trade.price,
|
||
"amount": trade.amount,
|
||
# TODO(BIQ-38115): что из «Объем»/«Расчетная сумма» — сумма с НКД, Q17.
|
||
"amount_with_accrued_interest": None,
|
||
"settlement_amount": trade.settlement_amount,
|
||
"comment": trade.comment or None,
|
||
"settlement_code": trade.settlement_code or None,
|
||
"order_number": trade.order_number or None,
|
||
"trade_number": None, # в выгрузке нет, Q2
|
||
"counterparty": trade.counterparty or None,
|
||
"commission_ts": trade.commission_ts,
|
||
"commission_clearing": trade.commission_clearing,
|
||
"commission_exchange": trade.commission_exchange,
|
||
"commission_tc": trade.commission_tc,
|
||
"commission_broker": trade.commission_broker,
|
||
"import_batch_id": batch_id,
|
||
"source_row": trade.ordinal,
|
||
"created_by": user,
|
||
"updated_by": user,
|
||
}
|
||
|
||
|
||
def build_plan(
|
||
trades: list[QuikTrade],
|
||
context: ImportContext,
|
||
portfolio_code: str,
|
||
decisions: Mapping[int, str],
|
||
) -> ImportPlan:
|
||
"""decisions: ordinal строки без портфеля -> INCLUDE / EXCLUDE."""
|
||
loaded_in: dict[int, str] = {}
|
||
for batch in context.same_file:
|
||
for ordinal in batch.loaded_rows:
|
||
loaded_in.setdefault(ordinal, batch.portfolio_code)
|
||
|
||
rows = []
|
||
for trade in trades:
|
||
hint = portfolio_hint(trade.comment)
|
||
where = loaded_in.get(trade.ordinal)
|
||
if where == portfolio_code:
|
||
status = RowStatus.LOADED_HERE
|
||
elif where is not None:
|
||
status = RowStatus.LOADED_ELSEWHERE
|
||
elif hint == portfolio_code:
|
||
status = RowStatus.OWN
|
||
elif hint is not None:
|
||
status = RowStatus.OTHER
|
||
elif decisions.get(trade.ordinal) == INCLUDE:
|
||
status = RowStatus.UNKNOWN_INCLUDED
|
||
elif decisions.get(trade.ordinal) == EXCLUDE:
|
||
status = RowStatus.UNKNOWN_EXCLUDED
|
||
else:
|
||
status = RowStatus.UNKNOWN_PENDING
|
||
rows.append(
|
||
PlannedRow(
|
||
trade=trade,
|
||
status=status,
|
||
hint=hint,
|
||
loaded_in=where,
|
||
securities=tuple(context.securities.get(trade.instrument, ())),
|
||
)
|
||
)
|
||
|
||
duplicate = next(
|
||
(b for b in context.same_file if b.portfolio_code == portfolio_code), None
|
||
)
|
||
return ImportPlan(
|
||
portfolio_code=portfolio_code,
|
||
rows=rows,
|
||
duplicate=duplicate,
|
||
overlapping=[
|
||
b for b in context.overlapping if b.portfolio_code == portfolio_code
|
||
],
|
||
summary=file_summary(len(trades), context),
|
||
)
|
||
|
||
|
||
def file_summary(rows_total: int, context: ImportContext) -> FileSummary:
|
||
loaded = {p.code: 0 for p in context.portfolios}
|
||
everywhere: set[int] = set()
|
||
for batch in context.same_file:
|
||
loaded[batch.portfolio_code] = (
|
||
loaded.get(batch.portfolio_code, 0) + batch.rows_loaded
|
||
)
|
||
everywhere |= batch.loaded_rows
|
||
return FileSummary(
|
||
rows_total=rows_total, loaded=loaded, nowhere=rows_total - len(everywhere)
|
||
)
|
||
|
||
|
||
def totals(trades: Iterable[QuikTrade]) -> list[TotalRow]:
|
||
"""Итоги по операции и валюте — валюты не смешиваем."""
|
||
groups: dict[tuple[str, str], list[QuikTrade]] = defaultdict(list)
|
||
for trade in trades:
|
||
groups[(trade.operation, trade.currency)].append(trade)
|
||
return [
|
||
TotalRow(
|
||
operation=operation,
|
||
currency=currency,
|
||
trades=len(group),
|
||
quantity=sum((t.quantity for t in group), Decimal(0)),
|
||
amount=sum((t.amount for t in group), Decimal(0)),
|
||
settlement_amount=sum((t.settlement_amount for t in group), Decimal(0)),
|
||
)
|
||
for (operation, currency), group in sorted(groups.items())
|
||
]
|
||
|
||
|
||
def commission_totals(trades: Iterable[QuikTrade]) -> list[CommissionTotal]:
|
||
"""Пять комиссий по отдельности. Складывать их между собой нельзя:
|
||
ТС = клиринговая + ФБ (D7)."""
|
||
groups: dict[str, list[QuikTrade]] = defaultdict(list)
|
||
for trade in trades:
|
||
groups[trade.currency].append(trade)
|
||
names = (
|
||
"commission_ts",
|
||
"commission_clearing",
|
||
"commission_exchange",
|
||
"commission_tc",
|
||
"commission_broker",
|
||
)
|
||
return [
|
||
CommissionTotal(
|
||
currency,
|
||
*(sum((getattr(t, name) for t in group), Decimal(0)) for name in names),
|
||
)
|
||
for currency, group in sorted(groups.items())
|
||
]
|
||
|
||
|
||
def _batch_info(batch: ImportBatch, portfolio_code: str) -> BatchInfo:
|
||
return BatchInfo(
|
||
portfolio_code=portfolio_code,
|
||
file_name=batch.file_name,
|
||
date_from=batch.date_from,
|
||
date_to=batch.date_to,
|
||
rows_loaded=batch.rows_loaded,
|
||
rows_excluded=batch.rows_excluded,
|
||
loaded_rows=frozenset(batch.loaded_rows),
|
||
created_at=batch.created_at,
|
||
created_by=batch.created_by,
|
||
)
|