"""The universal event-CSV parser, checked against the real Snowball export. The fixture is the only file in the project that covers every brokerage account at once, so these tests double as its documentation: the exact number of rows of each type, which of them deliberately never become ledger events, and what the format's ambiguities were decided to mean. A change in any of those numbers is a change in the import contract, not a test that needs relaxing. """ from __future__ import annotations import collections import dataclasses from decimal import Decimal from pathlib import Path import pytest from fintracker.models.ledger import EventKind from fintracker.sources.reports.base import BrokerEvent, ParsedReport, fingerprint_key from fintracker.sources.reports.csv_universal import UniversalCsvParser FIXTURES = Path(__file__).resolve().parents[1] / "fixtures" / "reports" SNOWBALL = FIXTURES / "snowball" HEADER = ( "Event,Date,Symbol,Price,Quantity,Currency,FeeTax,Exchange,NKD,FeeCurrency,DoNotAdjustCash,Note" ) #: What the fixture must yield, per kind. Sums to 535 of its 545 rows: 5 CUSTOM_HOLDING_PRICE #: and 1 CUSTOM_HOLDING_SETTINGS are not events at all, and 4 cash rows are Snowball's own #: balance corrections. EXPECTED_COUNTS = { EventKind.buy: 270, EventKind.sell: 60, EventKind.dividend: 80, EventKind.deposit: 96, EventKind.withdrawal: 5, EventKind.amortization: 9, EventKind.commission: 7, EventKind.tax: 4, EventKind.tax_refund: 2, EventKind.stock_split: 2, } @pytest.fixture(scope="module") def parser() -> UniversalCsvParser: return UniversalCsvParser() @pytest.fixture(scope="module") def fixture_path() -> Path: files = sorted(SNOWBALL.glob("*.csv")) assert files, f"нет фикстуры в {SNOWBALL}" return files[0] @pytest.fixture(scope="module") def report(parser: UniversalCsvParser, fixture_path: Path) -> ParsedReport: return parser.parse(fixture_path.read_bytes(), fixture_path.name) def _csv(*rows: str) -> bytes: return ("" + "\n".join((HEADER, *rows)) + "\n").encode("utf-8") def _key(report: ParsedReport, event: BrokerEvent) -> str: """The dedupe key `ledger/ingest.py` will build for this event (plan §1.6 A).""" return fingerprint_key( report.broker, report.account_external_id, event.kind, event.instrument.key() if event.instrument else "", event.trade_date, event.quantity, event.price, event.currency, amount=event.amount, seq=event.seq, ) def _floats(value: object, path: str) -> list[str]: """Every float reachable from `value`, by path — money must be Decimal end to end.""" if isinstance(value, bool | str | bytes) or value is None: return [] if isinstance(value, float): return [f"{path} = {value!r}"] if dataclasses.is_dataclass(value) and not isinstance(value, type): return [ bad for f in dataclasses.fields(value) for bad in _floats(getattr(value, f.name), f"{path}.{f.name}") ] if isinstance(value, dict): return [bad for k, v in value.items() for bad in _floats(v, f"{path}[{k!r}]")] if isinstance(value, list | tuple | set): return [bad for i, v in enumerate(value) for bad in _floats(v, f"{path}[{i}]")] return [] # --- 1. sniff ------------------------------------------------------------------------- def test_sniff_recognises_the_fixture(parser: UniversalCsvParser, fixture_path: Path) -> None: assert parser.sniff(fixture_path.read_bytes(), fixture_path.name) is True def test_sniff_ignores_the_filename(parser: UniversalCsvParser) -> None: """Recognition is by the header, so a renamed export is still recognised...""" data = _csv('"CASH_IN","2025-02-26 09:22:54","RUB","1","5000","RUB","0","","","","False",""') assert parser.sniff(data, "какой-то-файл.csv") is True assert parser.sniff(data, "") is True def test_sniff_rejects_other_report_formats(parser: UniversalCsvParser) -> None: """...and a file that merely has an extension we read is not ours.""" others = sorted((FIXTURES / "sber").glob("*.html")) + sorted((FIXTURES / "vtb").glob("*.xlsx")) assert others, "нет чужих фикстур для отрицательной проверки" for path in others: assert parser.sniff(path.read_bytes(), path.name) is False, path.name def test_sniff_rejects_a_csv_with_foreign_columns(parser: UniversalCsvParser) -> None: assert parser.sniff(b"a,b,c\n1,2,3\n", "x.csv") is False assert parser.sniff(b"", "x.csv") is False assert parser.sniff(b"\x00\x01\x02", "x.csv") is False def test_parse_rejects_a_file_it_cannot_read(parser: UniversalCsvParser) -> None: from fintracker.sources.reports.base import ParseError with pytest.raises(ParseError): parser.parse(b"a,b,c\n1,2,3\n", "x.csv") with pytest.raises(ParseError): parser.parse(_csv(), "empty.csv") # --- 2. what the fixture yields -------------------------------------------------------- def test_report_header(report: ParsedReport) -> None: import datetime assert report.broker == "csv" assert report.parser_version == "1" # the file names a portfolio, never an account — see the warning below assert report.account_external_id == "Мой капитал" assert report.period_from == datetime.date(2025, 2, 26) assert report.period_to == datetime.date(2026, 9, 17) def test_event_counts_by_kind(report: ParsedReport) -> None: counts = collections.Counter(e.kind for e in report.events) assert dict(counts) == EXPECTED_COUNTS assert len(report.events) == 535 assert report.meta["row_count"] == 545 def test_instruments_are_deduplicated(report: ParsedReport) -> None: keys = [ref.key() for ref in report.instruments] assert len(keys) == len(set(keys)) == 49 # bonds arrive by ISIN, everything else by ticker; cash rows name no instrument assert "ISIN:RU000A10B7T7" in keys assert "TICKER:SBER" in keys assert "TICKER:SU26212RMFS9" in keys, "секунда ОФЗ — не ISIN, а тикер MOEX" assert not any(k.startswith("TICKER:RUB") for k in keys) # --- 3. decimals ---------------------------------------------------------------------- def test_decimal_comma_is_parsed(parser: UniversalCsvParser) -> None: data = _csv( '"BUY","2025-03-01 10:00:00","SBER","4,48","219","RUB","6,57","MCX","0","RUB","False",""' ) (event,) = parser.parse(data, "renamed.csv").events assert event.price == Decimal("4.48") assert event.quantity == Decimal("219") assert event.fee == Decimal("6.57") assert event.amount == Decimal("-987.69") # 4.48 * 219 + 6.57, fee capitalised def test_no_floats_anywhere_in_the_output(report: ParsedReport) -> None: bad = _floats(report, "report") assert bad == [] def test_empty_numeric_cell_is_none_not_zero(parser: UniversalCsvParser) -> None: """An absent НКД is not a zero НКД: one is «not a bond», the other is «no interest».""" data = _csv('"BUY","2025-03-01 10:00:00","SBER","100","1","RUB","0","MCX","","","False",""') (event,) = parser.parse(data, "x.csv").events assert event.accrued_interest is None assert event.fee == Decimal(0) def test_bond_trade_adds_accrued_interest_to_the_money(report: ParsedReport) -> None: """НКД is paid on top of the price by the buyer — it must be inside `amount`.""" buy = next( e for e in report.events if e.kind is EventKind.buy and e.instrument is not None and e.instrument.isin == "RU000A107AM4" and e.accrued_interest == Decimal("5.08") ) assert buy.quantity == Decimal(4) assert buy.price == Decimal("929.25") assert buy.amount == -(Decimal(4) * Decimal("929.25") + Decimal("5.08") + Decimal("11.15")) # --- 4. dividends and coupons ---------------------------------------------------------- def test_dividend_money_lands_in_amount_not_quantity(report: ParsedReport) -> None: dividends = [e for e in report.events if e.kind is EventKind.dividend] assert len(dividends) == 80 assert all(e.quantity is None for e in dividends), "Quantity у DIVIDEND — это деньги" assert all(e.price is None for e in dividends) assert all(e.amount > 0 for e in dividends) assert sum(e.amount for e in dividends) == Decimal("12949.07") def test_bond_dividends_carry_the_coupon_hint(report: ParsedReport) -> None: """The parser cannot know an asset class, so it hands ingest a suspicion, not a kind.""" dividends = [e for e in report.events if e.kind is EventKind.dividend] hinted = [e for e in dividends if e.meta.get("payout_hint") == "coupon"] assert len(hinted) == 44 assert all( e.instrument is not None and ( (e.instrument.isin or "").startswith("RU000A") or (e.instrument.ticker or "").startswith("SU") ) for e in hinted ) # a share payout must never be hinted assert all( e.meta.get("payout_hint") is None for e in dividends if e.instrument is not None and e.instrument.ticker in {"T", "ROSN", "SBERP", "TATNP"} ) # --- 5. splits ------------------------------------------------------------------------- def test_split_ratio_and_collapsing_dedupe_key(report: ParsedReport) -> None: """Two identical SPLIT rows for T are one corporate action seen on two accounts. The format has no account column, so both rows land on the single account the user points the import at. They therefore share a `fingerprint_key` on purpose: replaying the ratio twice would multiply the position by 100 instead of by 10. This test pins that behaviour — if a future format gains an account column, it is expected to break. """ splits = [e for e in report.events if e.kind is EventKind.stock_split] assert len(splits) == 2 assert {e.meta["split_ratio"] for e in splits} == {Decimal(10)} assert {e.meta["ratio"] for e in splits} == {Decimal(10)} assert all(e.amount == Decimal(0) and e.quantity is None for e in splits) assert all(e.instrument is not None and e.instrument.ticker == "T" for e in splits) assert [e.seq for e in splits] == [0, 0] assert _key(report, splits[0]) == _key(report, splits[1]) def test_every_other_event_keeps_a_distinct_dedupe_key(report: ParsedReport) -> None: """The split pair is the ONLY intentional collapse in the file. In particular the pairs of equal same-day top-ups (two 1 100 ₽ one second apart, two 4 000 ₽ that differ only by their note) survive as distinct events: they are real money into two accounts, and netting them would understate the XIRR denominator. """ keys = [_key(report, e) for e in report.events] collapsed = [k for k, n in collections.Counter(keys).items() if n > 1] assert len(collapsed) == 1 assert len(set(keys)) == 534 same_second = [e for e in report.events if e.kind is EventKind.deposit and e.seq > 0] assert same_second, "в файле есть одинаковые пополнения одного дня — они должны выживать" def test_seq_is_stable_across_reparsing(parser: UniversalCsvParser, fixture_path: Path) -> None: """A re-import of the same bytes must reproduce every key, or nothing ever upserts.""" data = fixture_path.read_bytes() first = parser.parse(data, fixture_path.name) second = parser.parse(data, "совсем другое имя.csv") assert [_key(first, e) for e in first.events] == [_key(first, e) for e in second.events] # --- 6. amortisation ------------------------------------------------------------------- def test_amortisation_is_money_on_a_bond(report: ParsedReport) -> None: amortisations = [e for e in report.events if e.kind is EventKind.amortization] assert len(amortisations) == 9 assert all(e.amount > 0 for e in amortisations) assert all(e.quantity is None and e.price is None for e in amortisations) assert {e.instrument.isin for e in amortisations if e.instrument} == {"RU000A10B7T7"} assert all( e.instrument is not None and e.instrument.asset_class_hint == "bond" for e in amortisations ) assert sum(e.amount for e in amortisations) == Decimal("4530.07") # --- 7. manual prices ------------------------------------------------------------------ def test_custom_holding_prices_go_to_meta_not_to_events(report: ParsedReport) -> None: """CUSTOM_HOLDING_PRICE is a `price_manual` row, not a ledger event.""" prices = report.meta["manual_prices"] assert len(prices) == 5 assert {p["instrument_key"] for p in prices} == {"TICKER:SIBN6P4"} assert {p["currency"] for p in prices} == {"RUB"} assert all(isinstance(p["price"], Decimal) for p in prices) assert max(p["price"] for p in prices) == Decimal("12320.94504") assert all(e.meta.get("csv_event") != "CUSTOM_HOLDING_PRICE" for e in report.events) # the instrument they price must be resolvable, so it is also listed assert "TICKER:SIBN6P4" in {ref.key() for ref in report.instruments} def test_the_export_has_no_price_for_the_other_unquoted_paper(report: ParsedReport) -> None: """NDM_TBNK-PP-FIXPRCNT-08.25 is simply absent — this file cannot price it.""" named = {(ref.ticker or "") + (ref.isin or "") for ref in report.instruments} assert not any("NDM" in name for name in named) assert not any("NDM" in p["instrument_key"] for p in report.meta["manual_prices"]) def test_custom_holding_settings_describe_an_instrument(report: ParsedReport) -> None: """Its Note is JSON with `"` rewritten to `@*@` and Cyrillic as \\uXXXX escapes.""" ref = next(r for r in report.instruments if r.ticker == "SIBN6P4") assert ref.name == "Газпром Нефть 006Р-04" assert ref.currency == "RUB" assert ref.meta["custom_holding"] is True assert ref.meta["sector"] == "Other" assert ref.meta["settings"]["CustomHoldingType"] == 0 assert all(e.meta.get("csv_event") != "CUSTOM_HOLDING_SETTINGS" for e in report.events) def test_broken_settings_json_is_a_warning_not_a_crash(parser: UniversalCsvParser) -> None: data = _csv( '"CUSTOM_HOLDING_SETTINGS","2026-09-17 00:00:00","XYZ","0","0","RUB","0",' '"CUSTOM_HOLDING","0","","","{@*@Holding@*@:' ) result = parser.parse(data, "x.csv") assert result.events == [] assert any("не удалось разобрать JSON" in w for w in result.warnings) assert [ref.key() for ref in result.instruments] == ["TICKER:XYZ"] # --- 8. balance adjustments ------------------------------------------------------------ def test_balance_adjustments_never_become_flows(report: ParsedReport) -> None: """Snowball's own corrections would feed XIRR an external flow that never happened.""" adjustments = report.meta["balance_adjustments"] assert len(adjustments) == 4 assert collections.Counter(a["kind"] for a in adjustments) == {"CASH_OUT": 3, "CASH_IN": 1} assert all(isinstance(a["amount"], Decimal) for a in adjustments) # detection is by the note, not by the amount: one correction is 1017,84 ₽ while a # genuine withdrawal in the same file is 32 ₽ assert min(abs(a["amount"]) for a in adjustments) == Decimal("0.0024") assert max(abs(a["amount"]) for a in adjustments) == Decimal("1017.83999543") excluded = {(a["kind"], a["d"], abs(a["amount"])) for a in adjustments} for event in report.events: kind = "CASH_IN" if event.kind is EventKind.deposit else "CASH_OUT" assert (kind, event.trade_date, abs(event.amount)) not in excluded assert any("правка остатка" in w for w in report.warnings) def test_real_cash_flows_survive(report: ParsedReport) -> None: deposits = [e for e in report.events if e.kind is EventKind.deposit] withdrawals = [e for e in report.events if e.kind is EventKind.withdrawal] assert len(deposits) == 96 and len(withdrawals) == 5 assert all(e.amount > 0 and e.instrument is None for e in deposits) assert all(e.amount < 0 and e.instrument is None for e in withdrawals) assert sum(e.amount for e in withdrawals) == -Decimal("51294.41") assert sum(e.amount for e in deposits) == Decimal("454058.45") # --- 9. fees, taxes, refunds ----------------------------------------------------------- def test_fee_tax_and_refund_signs(report: ParsedReport) -> None: fees = [e for e in report.events if e.kind is EventKind.commission] taxes = [e for e in report.events if e.kind is EventKind.tax] refunds = [e for e in report.events if e.kind is EventKind.tax_refund] assert (len(fees), len(taxes), len(refunds)) == (7, 4, 2) assert all(e.instrument is None for e in fees + taxes + refunds) assert all(e.quantity is None and e.price is None for e in fees + taxes + refunds) assert all(e.amount == Decimal(-45) and e.fee == Decimal(45) for e in fees) assert [e.amount for e in taxes] == [Decimal(-234), Decimal(-3), Decimal(-22), Decimal(-32)] assert [e.amount for e in refunds] == [Decimal(3), Decimal(32)] assert all(e.tax == -e.amount for e in taxes) assert all(e.tax == e.amount for e in refunds) # --- unknown types and reconciliation warnings ----------------------------------------- def test_unknown_event_type_is_skipped_with_a_warning(parser: UniversalCsvParser) -> None: """Never `EventKind.other`: a silent bucket hides a format that grew a new row type.""" data = _csv( '"MARGIN_CALL","2025-03-01 10:00:00","SBER","1","1","RUB","0","MCX","0","","False",""', '"BUY","2025-03-01 10:00:00","SBER","100","1","RUB","0","MCX","0","","False",""', ) result = parser.parse(data, "x.csv") assert [e.kind for e in result.events] == [EventKind.buy] assert result.meta["unknown_events"] == {"MARGIN_CALL": 1} assert any("MARGIN_CALL" in w for w in result.warnings) assert not any(e.kind is EventKind.other for e in result.events) def test_closing_balances_are_absent_and_said_to_be(report: ParsedReport) -> None: assert report.positions_end == [] assert report.cash_end == [] assert any("positions_end" in w for w in report.warnings) def test_warns_that_the_target_account_is_the_user_s_to_choose(report: ParsedReport) -> None: assert any("shadow" in w and "все брокерские счета" in w for w in report.warnings)