"""Парсер выгрузки QUIK: реальная фикстура + синтетические файлы, собранные openpyxl прямо в тесте.""" from __future__ import annotations import io from datetime import date, datetime, time from decimal import Decimal from pathlib import Path from typing import Any import pytest from openpyxl import Workbook from app.imports.quik import BUY, SELL, QuikFormatError, parse_quik_xlsx from tests.conftest import FIXTURES ANON_FIXTURE = "quik_sample_300326_anon.xlsx" ORIGINAL_FIXTURE = "quik_sample_300326.xlsx" SHEET2_HEADERS = [ None, "Дата сделки", "Время", "Дата расчетов", "Инструмент сокр.", "Операция", "Счет", "Кол-во", "Цена", "Объем", "Расчетная сумма", "Комментарий", "Код расчетов", "Заявка", "Орг-я партнера", "Инструмент сокр.", "Валюта расчетов", "Комиссия ТС", "Клиринговая комиссия", "ФБ комиссия", "ТЦ комиссия", "Комиссия брокера", ] SHEET1_HEADERS = [ None, "Дата сделки", "Время", "Дата расчетов", "Инструмент сокр.", "Операция", "Кол-во", "Цена", "Объем", "НКД", "Расчетная сумма", "Орг-я партнера", "Код расчетов", "Валюта", "Доходность", "Счет", ] GOOD_ROW: dict[str, Any] = { "Дата сделки": datetime(2026, 3, 30), "Время": time(13, 53, 10), "Дата расчетов": datetime(2026, 3, 30), "Инструмент сокр.": "ЕврХол3P06", "Операция": "Купля", "Счет": "S00+00000000", "Кол-во": 54784, "Цена": 100, "Объем": 54784000, "Расчетная сумма": 54784000, "Комментарий": "/pp", "Код расчетов": "Z0", "Заявка": 10000000001, "Орг-я партнера": "ВТБКапТрейд", "Валюта расчетов": "SUR", "Комиссия ТС": 8217.6, "Клиринговая комиссия": 3492.48, "ФБ комиссия": 4725.12, "ТЦ комиссия": 0, "Комиссия брокера": 0, } def _row(headers: list[str | None], values: dict[str, Any], number: int) -> list: row = [] seen_instrument = False for header in headers: if header is None: row.append(number) elif header == "Инструмент сокр." and seen_instrument: row.append(values.get("Инструмент сокр. (2)", values.get(header))) else: seen_instrument |= header == "Инструмент сокр." row.append(values.get(header)) return row def _xlsx( rows: list[dict[str, Any]], *, headers: list[str | None] = SHEET2_HEADERS, title: str = "Лист2", extra_sheets: dict[str, list[list[Any]]] | None = None, ) -> bytes: wb = Workbook() ws = wb.active ws.title = title ws.append(headers) for number, values in enumerate(rows, start=1): ws.append(_row(headers, values, number)) for name, sheet_rows in (extra_sheets or {}).items(): extra = wb.create_sheet(name) for sheet_row in sheet_rows: extra.append(sheet_row) buffer = io.BytesIO() wb.save(buffer) return buffer.getvalue() def _with(**changes: Any) -> dict[str, Any]: row = dict(GOOD_ROW) row.update(changes) return row # --- реальная фикстура ------------------------------------------------------ def test_real_fixture_sheet2(fixtures_dir: Path) -> None: result = parse_quik_xlsx((fixtures_dir / ANON_FIXTURE).read_bytes()) assert result.sheet == "Лист2" # «Лист1» в том же файле пропущен assert result.ok assert result.errors == [] and result.warnings == [] assert len(result.trades) == 6 assert [t.operation for t in result.trades] == [BUY, BUY, BUY, SELL, BUY, SELL] assert (result.date_from, result.date_to) == (date(2026, 3, 30),) * 2 first = result.trades[0] assert (first.ordinal, first.row) == (1, 2) assert first.trade_time == time(13, 53, 10) assert first.instrument == "ЕврХол3P06" assert first.currency == "RUB" assert first.comment == "/pp" assert first.order_number == "10000000001" assert first.quantity == Decimal("54784") assert first.price == Decimal("100") assert first.amount == Decimal("54784000") # D7: ТС = клиринговая + ФБ, Decimal без хвостов float. assert first.commission_ts == Decimal("8217.6") assert first.commission_clearing == Decimal("3492.48") assert first.commission_exchange == Decimal("4725.12") assert first.commission_ts == first.commission_clearing + first.commission_exchange assert result.trades[3].price == Decimal("99.85") assert sum(t.quantity for t in result.trades if t.operation == BUY) == 1098000 assert len(result.content_hash or "") == 64 def test_fixture_is_anonymized(fixtures_dir: Path) -> None: result = parse_quik_xlsx((fixtures_dir / ANON_FIXTURE).read_bytes()) assert {t.account for t in result.trades} == {"S00+00000000"} assert [t.counterparty for t in result.trades] == [ "Контрагент 1", "Контрагент 1", "Контрагент 2", "Контрагент 3", "Контрагент 2", "Контрагент 3", ] assert [t.order_number for t in result.trades] == [ str(10000000001 + i) for i in range(6) ] # Оригинал выгрузки в git не хранится (см. README); если он лежит рядом — # проверяем, что обезличенная копия отличается только счётом, контрагентами # и номерами заявок. ANON_FIELDS = {"account", "counterparty", "order_number"} @pytest.mark.skipif( not (FIXTURES / ORIGINAL_FIXTURE).exists(), reason=f"нет tests/fixtures/{ORIGINAL_FIXTURE}", ) def test_anonymized_fixture_matches_original(fixtures_dir: Path) -> None: original = parse_quik_xlsx((fixtures_dir / ORIGINAL_FIXTURE).read_bytes()) anon = parse_quik_xlsx((fixtures_dir / ANON_FIXTURE).read_bytes()) assert (anon.sheet, anon.date_from, anon.date_to) == ( original.sheet, original.date_from, original.date_to, ) assert len(anon.trades) == len(original.trades) for a, o in zip(anon.trades, original.trades, strict=True): assert {k: v for k, v in vars(a).items() if k not in ANON_FIELDS} == { k: v for k, v in vars(o).items() if k not in ANON_FIELDS } # Синтетические номера сохраняют, какие сделки шли одной заявкой. assert len({t.order_number for t in anon.trades}) == len( {t.order_number for t in original.trades} ) def test_only_sheet1_format_is_rejected_with_hint() -> None: sheet1_row = {"Валюта": "SUR", "НКД": None, "Доходность": 15.41, **GOOD_ROW} data = _xlsx([sheet1_row], headers=SHEET1_HEADERS, title="Лист1") with pytest.raises(QuikFormatError, match="коротком формате") as exc_info: parse_quik_xlsx(data) assert "Комментарий" in str(exc_info.value) assert "Лист2" in str(exc_info.value) # --- формат файла ----------------------------------------------------------- def test_not_xlsx_is_rejected() -> None: with pytest.raises(QuikFormatError, match="Не удалось открыть файл как .xlsx"): parse_quik_xlsx(b"\xd0\xcf\x11\xe0 old xls") def test_missing_columns_are_listed() -> None: headers = [h for h in SHEET2_HEADERS if h not in ("Заявка", "ТЦ комиссия")] with pytest.raises(QuikFormatError) as exc_info: parse_quik_xlsx(_xlsx([GOOD_ROW], headers=headers)) assert "нет колонок: «Заявка», «ТЦ комиссия»" in str(exc_info.value) def test_no_trade_table_at_all() -> None: wb = Workbook() wb.active.append(["Отчёт", "что-то другое"]) buffer = io.BytesIO() wb.save(buffer) with pytest.raises(QuikFormatError, match="Не найдена таблица сделок"): parse_quik_xlsx(buffer.getvalue()) def test_header_only_file_has_no_trades() -> None: with pytest.raises(QuikFormatError, match="нет ни одной сделки"): parse_quik_xlsx(_xlsx([])) def test_sheet_found_by_headers_not_name_with_title_rows_and_column_order() -> None: """Пересохранённый файл: лист «Сделки», шапка над таблицей, колонки в другом порядке, лишняя колонка — всё равно разбирается.""" headers = list(reversed(SHEET2_HEADERS[1:])) + ["Лишняя колонка"] wb = Workbook() ws = wb.active ws.title = "Сделки" ws.append(["Выгрузка из QUIK за 30.03.2026"]) ws.append([]) ws.append(headers) ws.append(_row(headers, {**GOOD_ROW, "Лишняя колонка": "x"}, 1)) buffer = io.BytesIO() wb.save(buffer) result = parse_quik_xlsx(buffer.getvalue()) assert result.sheet == "Сделки" assert result.ok assert result.trades[0].row == 4 assert result.trades[0].quantity == Decimal("54784") def test_empty_rows_are_skipped_and_do_not_change_hash() -> None: compact = parse_quik_xlsx(_xlsx([GOOD_ROW, _with(Операция="Продажа")])) wb = Workbook() ws = wb.active ws.append(SHEET2_HEADERS) ws.append(_row(SHEET2_HEADERS, GOOD_ROW, 1)) ws.append([None] * 5) ws.append(["", " "]) ws.append(_row(SHEET2_HEADERS, _with(Операция="Продажа"), 2)) for _ in range(30): ws.append([None] * len(SHEET2_HEADERS)) buffer = io.BytesIO() wb.save(buffer) sparse = parse_quik_xlsx(buffer.getvalue()) assert [t.ordinal for t in sparse.trades] == [1, 2] assert [t.row for t in sparse.trades] == [2, 5] assert sparse.content_hash == compact.content_hash # --- значения ячеек --------------------------------------------------------- def test_serial_numbers_fractions_and_text_values() -> None: row = _with( **{ "Дата сделки": 46111, # серийное число Excel = 30.03.2026 "Время": 0.5, # доля суток = 12:00:00 "Дата расчетов": "31.03.2026", "Кол-во": "1 000", "Цена": "99,85", "Объем": "998 500,00", "Расчетная сумма": "998500", "Заявка": "10000000001", "Комиссия ТС": "15,5", "Клиринговая комиссия": "10,5", "ФБ комиссия": "5", "ТЦ комиссия": None, "Комиссия брокера": "", "Валюта расчетов": "cny", "Операция": "ПРОДАЖА", } ) result = parse_quik_xlsx(_xlsx([row])) assert result.errors == [], [str(e) for e in result.errors] [trade] = result.trades assert trade.trade_date == date(2026, 3, 30) assert trade.trade_time == time(12, 0, 0) assert trade.settlement_date == date(2026, 3, 31) assert trade.quantity == Decimal("1000") assert trade.price == Decimal("99.85") assert trade.amount == Decimal("998500.00") assert trade.currency == "CNY" assert trade.operation == SELL assert trade.commission_tc == Decimal("0") assert trade.commission_broker == Decimal("0") def test_text_time_and_iso_date() -> None: result = parse_quik_xlsx( _xlsx([_with(**{"Время": "17:04:36", "Дата сделки": "2026-03-30"})]) ) assert result.trades[0].trade_time == time(17, 4, 36) assert result.trades[0].trade_date == date(2026, 3, 30) def test_same_data_in_different_cell_types_gives_same_hash() -> None: as_numbers = parse_quik_xlsx(_xlsx([GOOD_ROW])) as_text = parse_quik_xlsx( _xlsx( [ _with( **{ "Цена": "100,00", "Кол-во": "54784", "Комиссия ТС": "8217,60", "Заявка": "10000000001", } ) ] ) ) assert as_numbers.content_hash == as_text.content_hash def test_different_data_gives_different_hash() -> None: one = parse_quik_xlsx(_xlsx([GOOD_ROW])) other = parse_quik_xlsx(_xlsx([_with(**{"Цена": 100.01})])) assert one.content_hash != other.content_hash # --- ошибки и предупреждения по строкам ---------------------------------------- def test_row_errors_are_collected_with_excel_row_and_column() -> None: rows = [ GOOD_ROW, _with(**{"Дата сделки": "32.03.2026", "Операция": "Обмен"}), _with(**{"Кол-во": -5, "Цена": "abc"}), _with(**{"Кол-во": 1.5, "Валюта расчетов": "EUR"}), _with(**{"Инструмент сокр.": None, "Время": "25:00"}), _with(**{"Комиссия ТС": -1, "Расчетная сумма": 0}), ] result = parse_quik_xlsx(_xlsx(rows)) assert not result.ok assert result.content_hash is None assert [t.row for t in result.trades] == [2] messages = [str(e) for e in result.errors] assert messages == [ "строка 3, «Дата сделки»: не дата: «32.03.2026» (ожидается ДД.ММ.ГГГГ)", "строка 3, «Операция»: неизвестная операция «Обмен» (ожидается Купля/Продажа)", "строка 4, «Цена»: не число: «abc»", "строка 4, «Кол-во»: должно быть больше нуля, а не -5", "строка 5, «Валюта расчетов»: валюта «EUR» не поддерживается " "(ожидается SUR/RUB, CNY, USD)", "строка 5, «Кол-во»: должно быть целым, а не 1.5", "строка 6, «Время»: не время: «25:00» (ожидается ЧЧ:ММ:СС)", "строка 6, «Инструмент сокр.»: пусто", "строка 7, «Расчетная сумма»: должно быть больше нуля, а не 0", "строка 7, «Комиссия ТС»: не может быть отрицательной: -1", ] def test_out_of_range_serial_date_is_an_error() -> None: result = parse_quik_xlsx(_xlsx([_with(**{"Дата сделки": 5})])) assert "дата вне допустимого диапазона" in str(result.errors[0]) def test_warnings_do_not_block() -> None: rows = [ _with(**{"Комиссия ТС": 100, "Клиринговая комиссия": 30, "ФБ комиссия": 20}), _with(**{"Дата расчетов": datetime(2026, 3, 27)}), _with(**{"Инструмент сокр. (2)": "Другое имя"}), ] result = parse_quik_xlsx(_xlsx(rows)) assert result.ok assert len(result.trades) == 3 assert [str(w) for w in result.warnings] == [ "строка 2, «Комиссия ТС»: 100 ≠ клиринговая + ФБ (50), см. D7", "строка 3, «Дата расчетов»: дата расчётов раньше даты сделки", "строка 4, «Инструмент сокр.»: во второй колонке «Инструмент сокр.» " "другое значение: Другое имя", ]