From c55fe19e4891083c5bc3e3a7afcfc7a79d8162dc Mon Sep 17 00:00:00 2001 From: Dmitry Date: Fri, 18 Sep 2026 13:43:49 +0300 Subject: [PATCH] =?UTF-8?q?feat(sources):=20=D0=BA=D0=BE=D0=BD=D1=82=D1=80?= =?UTF-8?q?=D0=B0=D0=BA=D1=82=20=D0=B8=D1=81=D1=82=D0=BE=D1=87=D0=BD=D0=B8?= =?UTF-8?q?=D0=BA=D0=BE=D0=B2,=20worker=20=D0=B8=20=D1=81=D0=B8=D0=BD?= =?UTF-8?q?=D0=BA=20ZenMoney=20+=20=D0=A6=D0=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Source.sync(ctx) -> SyncResult пишет только raw_* и возвращает курсор; локи, журнал, ошибки и продвижение курсора берёт на себя worker/runner. ZenMoney читается единственным доступным способом — POST /v8/diff/ по serverTimestamp; токен живёт сутки, поэтому worker ротирует refresh_token через source_credential. Маппер всегда пересобирает core из полных raw_*, так что удаление в ZenMoney исчезает и у нас. ЦБ ходит мимо прокси (trust_env=False) и отдаёт cp1251 с делением на Nominal. Курсы только по рабочим дням — протяжку по календарю делает аналитика. Планировщик — APScheduler в отдельном процессе, на источник advisory-лок sync:, чтобы ручной запуск не пересёкся с плановым. --- ...0506c15002c8_phase1_zenmoney_fx_metrics.py | 458 ++++++++++++++++ .../b1d54240abb8_account_balance_columns.py | 42 ++ backend/src/fintracker/models/pricing.py | 168 ++++++ backend/src/fintracker/models/zenmoney.py | 204 +++++++ backend/src/fintracker/sources/__init__.py | 13 + backend/src/fintracker/sources/base.py | 39 ++ .../src/fintracker/sources/cbr/__init__.py | 8 + backend/src/fintracker/sources/cbr/client.py | 163 ++++++ backend/src/fintracker/sources/cbr/sync.py | 161 ++++++ backend/src/fintracker/sources/registry.py | 24 + .../fintracker/sources/zenmoney/__init__.py | 8 + .../src/fintracker/sources/zenmoney/client.py | 280 ++++++++++ .../src/fintracker/sources/zenmoney/mapper.py | 483 ++++++++++++++++ .../src/fintracker/sources/zenmoney/sync.py | 169 ++++++ backend/src/fintracker/worker/__init__.py | 0 backend/src/fintracker/worker/jobs.py | 39 ++ backend/src/fintracker/worker/locks.py | 36 ++ backend/src/fintracker/worker/runner.py | 95 ++++ backend/src/fintracker/worker/scheduler.py | 118 ++++ backend/tests/fixtures/cbr/daily.xml | 27 + backend/tests/fixtures/cbr/dynamic_jpy.xml | 13 + backend/tests/fixtures/cbr/dynamic_usd.xml | 18 + backend/tests/fixtures/reports/raw/.gitkeep | 0 .../fixtures/zenmoney/diff_deletion.json | 30 + .../tests/fixtures/zenmoney/diff_empty.json | 4 + .../tests/fixtures/zenmoney/diff_full.json | 266 +++++++++ backend/tests/sources/__init__.py | 0 backend/tests/sources/conftest.py | 62 +++ backend/tests/sources/test_cbr.py | 153 ++++++ backend/tests/sources/test_zenmoney.py | 517 ++++++++++++++++++ backend/tests/test_sync.py | 115 ++++ 31 files changed, 3713 insertions(+) create mode 100644 backend/alembic/versions/0506c15002c8_phase1_zenmoney_fx_metrics.py create mode 100644 backend/alembic/versions/b1d54240abb8_account_balance_columns.py create mode 100644 backend/src/fintracker/models/pricing.py create mode 100644 backend/src/fintracker/models/zenmoney.py create mode 100644 backend/src/fintracker/sources/__init__.py create mode 100644 backend/src/fintracker/sources/base.py create mode 100644 backend/src/fintracker/sources/cbr/__init__.py create mode 100644 backend/src/fintracker/sources/cbr/client.py create mode 100644 backend/src/fintracker/sources/cbr/sync.py create mode 100644 backend/src/fintracker/sources/registry.py create mode 100644 backend/src/fintracker/sources/zenmoney/__init__.py create mode 100644 backend/src/fintracker/sources/zenmoney/client.py create mode 100644 backend/src/fintracker/sources/zenmoney/mapper.py create mode 100644 backend/src/fintracker/sources/zenmoney/sync.py create mode 100644 backend/src/fintracker/worker/__init__.py create mode 100644 backend/src/fintracker/worker/jobs.py create mode 100644 backend/src/fintracker/worker/locks.py create mode 100644 backend/src/fintracker/worker/runner.py create mode 100644 backend/src/fintracker/worker/scheduler.py create mode 100644 backend/tests/fixtures/cbr/daily.xml create mode 100644 backend/tests/fixtures/cbr/dynamic_jpy.xml create mode 100644 backend/tests/fixtures/cbr/dynamic_usd.xml create mode 100644 backend/tests/fixtures/reports/raw/.gitkeep create mode 100644 backend/tests/fixtures/zenmoney/diff_deletion.json create mode 100644 backend/tests/fixtures/zenmoney/diff_empty.json create mode 100644 backend/tests/fixtures/zenmoney/diff_full.json create mode 100644 backend/tests/sources/__init__.py create mode 100644 backend/tests/sources/conftest.py create mode 100644 backend/tests/sources/test_cbr.py create mode 100644 backend/tests/sources/test_zenmoney.py create mode 100644 backend/tests/test_sync.py diff --git a/backend/alembic/versions/0506c15002c8_phase1_zenmoney_fx_metrics.py b/backend/alembic/versions/0506c15002c8_phase1_zenmoney_fx_metrics.py new file mode 100644 index 0000000..19bbf9d --- /dev/null +++ b/backend/alembic/versions/0506c15002c8_phase1_zenmoney_fx_metrics.py @@ -0,0 +1,458 @@ +"""phase1 zenmoney fx metrics + +Revision ID: 0506c15002c8 +Revises: 05beabc436ad +Create Date: 2026-09-17 22:38:27.588642 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0506c15002c8" +down_revision: str | None = "05beabc436ad" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "category", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("source", sa.String(length=32), nullable=False), + sa.Column("source_id", sa.String(length=64), nullable=False), + sa.Column("parent_id", sa.Integer(), nullable=True), + sa.Column("name", sa.String(length=256), nullable=False), + sa.Column("icon", sa.String(length=64), nullable=True), + sa.Column("color", sa.BigInteger(), nullable=True), + sa.Column("show_income", sa.Boolean(), nullable=False), + sa.Column("show_outcome", sa.Boolean(), nullable=False), + sa.Column("archived", sa.Boolean(), nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.ForeignKeyConstraint( + ["parent_id"], + ["category.id"], + name=op.f("fk_category_parent_id_category"), + ondelete="SET NULL", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_category")), + sa.UniqueConstraint("source", "source_id", name=op.f("uq_category_source_source_id")), + ) + op.create_table( + "fx_rate_daily", + sa.Column("d", sa.Date(), nullable=False), + sa.Column("ccy", sa.String(length=3), nullable=False), + sa.Column("rate_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("source", sa.String(length=16), nullable=False), + sa.Column("is_carried", sa.Boolean(), nullable=False), + sa.PrimaryKeyConstraint("d", "ccy", name=op.f("pk_fx_rate_daily")), + ) + op.create_table( + "merchant", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("source", sa.String(length=32), nullable=False), + sa.Column("source_id", sa.String(length=64), nullable=False), + sa.Column("name", sa.String(length=256), nullable=False), + sa.PrimaryKeyConstraint("id", name=op.f("pk_merchant")), + sa.UniqueConstraint("source", "source_id", name=op.f("uq_merchant_source_source_id")), + ) + op.create_table( + "metric_cash_flow_monthly", + sa.Column("month", sa.Date(), nullable=False), + sa.Column("income_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("expense_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("baseline_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("one_off_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("savings_transfer_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("savings_rate", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("txn_count", sa.Integer(), nullable=False), + sa.Column( + "computed_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.PrimaryKeyConstraint("month", name=op.f("pk_metric_cash_flow_monthly")), + ) + op.create_table( + "metric_data_quality", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("check_name", sa.String(length=64), nullable=False), + sa.Column("severity", sa.String(length=8), nullable=False), + sa.Column("detail", sa.Text(), nullable=False), + sa.Column("count", sa.Integer(), nullable=False), + sa.Column("ref", postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column( + "computed_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_metric_data_quality")), + ) + op.create_index( + op.f("ix_metric_data_quality_check_name"), + "metric_data_quality", + ["check_name"], + unique=False, + ) + op.create_table( + "metric_net_worth_daily", + sa.Column("d", sa.Date(), nullable=False), + sa.Column("total_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("liquid_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("savings_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("investment_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("debt_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("by_currency", postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column("missing_fx_count", sa.Integer(), nullable=False), + sa.Column( + "computed_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.PrimaryKeyConstraint("d", name=op.f("pk_metric_net_worth_daily")), + ) + op.create_table( + "metric_refresh_log", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column( + "started_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("trigger", sa.String(length=32), nullable=False), + sa.Column("error", sa.Text(), nullable=True), + sa.PrimaryKeyConstraint("id", name=op.f("pk_metric_refresh_log")), + ) + op.create_table( + "metric_runway", + sa.Column("as_of", sa.Date(), nullable=False), + sa.Column("liquid_reserve_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("avg_baseline_3m_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("runway_months", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column( + "computed_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.PrimaryKeyConstraint("as_of", name=op.f("pk_metric_runway")), + ) + op.create_table( + "raw_cbr_rate", + sa.Column("rate_date", sa.Date(), nullable=False), + sa.Column("ccy", sa.String(length=3), nullable=False), + sa.Column("nominal", sa.Integer(), nullable=False), + sa.Column("value", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column( + "fetched_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.PrimaryKeyConstraint("rate_date", "ccy", name=op.f("pk_raw_cbr_rate")), + ) + op.create_table( + "raw_zenmoney_deletion", + sa.Column("entity_type", sa.String(length=32), nullable=False), + sa.Column("id", sa.String(length=64), nullable=False), + sa.Column("stamp", sa.BigInteger(), nullable=True), + sa.Column( + "deleted_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.PrimaryKeyConstraint("entity_type", "id", name=op.f("pk_raw_zenmoney_deletion")), + ) + op.create_table( + "raw_zenmoney_entity", + sa.Column("entity_type", sa.String(length=32), nullable=False), + sa.Column("id", sa.String(length=64), nullable=False), + sa.Column("changed", sa.BigInteger(), nullable=True), + sa.Column("payload", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column( + "ingested_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.PrimaryKeyConstraint("entity_type", "id", name=op.f("pk_raw_zenmoney_entity")), + ) + op.create_table( + "rule", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column( + "kind", + sa.Enum( + "savings", + "one_off", + "category", + "payee", + "broker_target", + "ignore", + name="rule_kind", + ), + nullable=False, + ), + sa.Column( + "match_type", + sa.Enum("id", "payee", "comment", "category", "mcc", "account", name="rule_match_type"), + nullable=False, + ), + sa.Column("pattern", sa.String(length=512), nullable=False), + sa.Column("value", sa.String(length=512), nullable=True), + sa.Column("note", sa.Text(), nullable=True), + sa.Column("enabled", sa.Boolean(), nullable=False), + sa.Column("priority", sa.Integer(), nullable=False), + sa.Column("last_matched_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("match_count", sa.Integer(), nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_rule")), + ) + op.create_table( + "trip", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("name", sa.String(length=256), nullable=False), + sa.Column("date_from", sa.Date(), nullable=False), + sa.Column("date_to", sa.Date(), nullable=False), + sa.Column("country", sa.String(length=2), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_trip")), + ) + op.create_table( + "cash_txn", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("source", sa.String(length=32), nullable=False), + sa.Column("source_id", sa.String(length=64), nullable=False), + sa.Column("ts", sa.DateTime(timezone=True), nullable=False), + sa.Column("date", sa.Date(), nullable=False), + sa.Column("income", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("income_currency", sa.String(length=3), nullable=True), + sa.Column("income_account_id", sa.Integer(), nullable=True), + sa.Column("outcome", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("outcome_currency", sa.String(length=3), nullable=True), + sa.Column("outcome_account_id", sa.Integer(), nullable=True), + sa.Column("op_income", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("op_income_currency", sa.String(length=3), nullable=True), + sa.Column("op_outcome", sa.Numeric(precision=24, scale=10), nullable=True), + sa.Column("op_outcome_currency", sa.String(length=3), nullable=True), + sa.Column("payee", sa.String(length=512), nullable=True), + sa.Column("original_payee", sa.String(length=512), nullable=True), + sa.Column("merchant_id", sa.Integer(), nullable=True), + sa.Column("comment", sa.Text(), nullable=True), + sa.Column("mcc", sa.Integer(), nullable=True), + sa.Column("hold", sa.Boolean(), nullable=False), + sa.Column("deleted", sa.Boolean(), nullable=False), + sa.Column("changed", sa.BigInteger(), nullable=True), + sa.Column("primary_category_id", sa.Integer(), nullable=True), + sa.Column( + "flow_type", + sa.Enum( + "income", + "expense", + "internal_transfer", + "savings_transfer", + "broker_external_flow", + "deleted", + "other", + name="flow_type", + ), + nullable=False, + ), + sa.Column("category_id", sa.Integer(), nullable=True), + sa.Column("payee_canonical", sa.String(length=512), nullable=True), + sa.Column("is_one_off", sa.Boolean(), nullable=False), + sa.Column("trip_id", sa.Integer(), nullable=True), + sa.Column("meta", postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.ForeignKeyConstraint( + ["category_id"], + ["category.id"], + name=op.f("fk_cash_txn_category_id_category"), + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["income_account_id"], + ["account.id"], + name=op.f("fk_cash_txn_income_account_id_account"), + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["merchant_id"], + ["merchant.id"], + name=op.f("fk_cash_txn_merchant_id_merchant"), + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["outcome_account_id"], + ["account.id"], + name=op.f("fk_cash_txn_outcome_account_id_account"), + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["primary_category_id"], + ["category.id"], + name=op.f("fk_cash_txn_primary_category_id_category"), + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["trip_id"], ["trip.id"], name=op.f("fk_cash_txn_trip_id_trip"), ondelete="SET NULL" + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_cash_txn")), + sa.UniqueConstraint("source", "source_id", name=op.f("uq_cash_txn_source_source_id")), + ) + op.create_index(op.f("ix_cash_txn_date"), "cash_txn", ["date"], unique=False) + op.create_index( + op.f("ix_cash_txn_income_account_id"), "cash_txn", ["income_account_id"], unique=False + ) + op.create_index( + op.f("ix_cash_txn_outcome_account_id"), "cash_txn", ["outcome_account_id"], unique=False + ) + op.create_table( + "metric_spending_by_category", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("month", sa.Date(), nullable=False), + sa.Column("category_id", sa.Integer(), nullable=True), + sa.Column("root_category_id", sa.Integer(), nullable=True), + sa.Column("amount_rub", sa.Numeric(precision=24, scale=10), nullable=False), + sa.Column("txn_count", sa.Integer(), nullable=False), + sa.Column( + "computed_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.ForeignKeyConstraint( + ["category_id"], + ["category.id"], + name=op.f("fk_metric_spending_by_category_category_id_category"), + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["root_category_id"], + ["category.id"], + name=op.f("fk_metric_spending_by_category_root_category_id_category"), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_metric_spending_by_category")), + sa.UniqueConstraint( + "month", + "category_id", + name=op.f("uq_metric_spending_by_category_month_category_id"), + postgresql_nulls_not_distinct=True, + ), + ) + op.create_index( + op.f("ix_metric_spending_by_category_month"), + "metric_spending_by_category", + ["month"], + unique=False, + ) + op.create_table( + "cash_txn_tag", + sa.Column("txn_id", sa.Integer(), nullable=False), + sa.Column("ord", sa.Integer(), nullable=False), + sa.Column("category_id", sa.Integer(), nullable=False), + sa.ForeignKeyConstraint( + ["category_id"], + ["category.id"], + name=op.f("fk_cash_txn_tag_category_id_category"), + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["txn_id"], + ["cash_txn.id"], + name=op.f("fk_cash_txn_tag_txn_id_cash_txn"), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("txn_id", "ord", name=op.f("pk_cash_txn_tag")), + ) + op.create_index( + op.f("ix_cash_txn_tag_category_id"), "cash_txn_tag", ["category_id"], unique=False + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index(op.f("ix_cash_txn_tag_category_id"), table_name="cash_txn_tag") + op.drop_table("cash_txn_tag") + op.drop_index( + op.f("ix_metric_spending_by_category_month"), table_name="metric_spending_by_category" + ) + op.drop_table("metric_spending_by_category") + op.drop_index(op.f("ix_cash_txn_outcome_account_id"), table_name="cash_txn") + op.drop_index(op.f("ix_cash_txn_income_account_id"), table_name="cash_txn") + op.drop_index(op.f("ix_cash_txn_date"), table_name="cash_txn") + op.drop_table("cash_txn") + op.drop_table("trip") + op.drop_table("rule") + op.drop_table("raw_zenmoney_entity") + op.drop_table("raw_zenmoney_deletion") + op.drop_table("raw_cbr_rate") + op.drop_table("metric_runway") + op.drop_table("metric_refresh_log") + op.drop_table("metric_net_worth_daily") + op.drop_index(op.f("ix_metric_data_quality_check_name"), table_name="metric_data_quality") + op.drop_table("metric_data_quality") + op.drop_table("metric_cash_flow_monthly") + op.drop_table("merchant") + op.drop_table("fx_rate_daily") + op.drop_table("category") + # ### end Alembic commands ### + for enum_name in ("flow_type", "rule_kind", "rule_match_type"): + sa.Enum(name=enum_name).drop(op.get_bind(), checkfirst=True) diff --git a/backend/alembic/versions/b1d54240abb8_account_balance_columns.py b/backend/alembic/versions/b1d54240abb8_account_balance_columns.py new file mode 100644 index 0000000..93d429b --- /dev/null +++ b/backend/alembic/versions/b1d54240abb8_account_balance_columns.py @@ -0,0 +1,42 @@ +"""account balance columns + +Revision ID: b1d54240abb8 +Revises: 0506c15002c8 +Create Date: 2026-09-17 22:40:21.874610 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "b1d54240abb8" +down_revision: str | None = "0506c15002c8" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.add_column( + "account", sa.Column("balance", sa.Numeric(precision=24, scale=10), nullable=True) + ) + op.add_column( + "account", sa.Column("start_balance", sa.Numeric(precision=24, scale=10), nullable=True) + ) + op.add_column( + "account", sa.Column("credit_limit", sa.Numeric(precision=24, scale=10), nullable=True) + ) + op.add_column("account", sa.Column("balance_as_of", sa.DateTime(timezone=True), nullable=True)) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_column("account", "balance_as_of") + op.drop_column("account", "credit_limit") + op.drop_column("account", "start_balance") + op.drop_column("account", "balance") + # ### end Alembic commands ### diff --git a/backend/src/fintracker/models/pricing.py b/backend/src/fintracker/models/pricing.py new file mode 100644 index 0000000..d8a9cd6 --- /dev/null +++ b/backend/src/fintracker/models/pricing.py @@ -0,0 +1,168 @@ +"""FX rates. Later phases add price_daily / price_last / corporate_action here (plan §1.6).""" + +from __future__ import annotations + +import enum +from datetime import date, datetime +from decimal import Decimal + +from sqlalchemy import Boolean, ForeignKey, Index, Integer, String, UniqueConstraint, func +from sqlalchemy.orm import Mapped, mapped_column + +from fintracker.db.base import Base, TimestampMixin, db_enum + + +class RawCbrRate(Base): + """CBR official rate as published: RUB per `nominal` units of `ccy`, on business days only.""" + + __tablename__ = "raw_cbr_rate" + + rate_date: Mapped[date] = mapped_column(primary_key=True) + ccy: Mapped[str] = mapped_column(String(3), primary_key=True) + nominal: Mapped[int] = mapped_column(Integer, default=1) + value: Mapped[Decimal] + """RUB per `nominal` units, exactly as CBR printed it.""" + fetched_at: Mapped[datetime] = mapped_column(server_default=func.now()) + + +class FxRateDaily(Base): + """Every calendar day has a rate: business-day quotes carried forward (and backward before + the first quote). RUB is present as 1.0. Rebuilt by pricing/fx.py on each refresh.""" + + __tablename__ = "fx_rate_daily" + + d: Mapped[date] = mapped_column(primary_key=True) + ccy: Mapped[str] = mapped_column(String(3), primary_key=True) + rate_rub: Mapped[Decimal] + """RUB per ONE unit of ccy.""" + source: Mapped[str] = mapped_column(String(16), default="cbr") + is_carried: Mapped[bool] = mapped_column(Boolean, default=False) + + +class PriceDaily(Base): + """Daily close per instrument. Bonds quote in percent of nominal, so `price_pct` carries + the quote as published and `close` is the money value it resolves to.""" + + __tablename__ = "price_daily" + + instrument_id: Mapped[int] = mapped_column( + ForeignKey("instrument.id", ondelete="CASCADE"), primary_key=True + ) + d: Mapped[date] = mapped_column(primary_key=True) + close: Mapped[Decimal] + open: Mapped[Decimal | None] + high: Mapped[Decimal | None] + low: Mapped[Decimal | None] + volume: Mapped[Decimal | None] + currency: Mapped[str] = mapped_column(String(3)) + source: Mapped[str] = mapped_column(String(16)) + """tinvest | moex | manual""" + price_pct: Mapped[Decimal | None] + """Bond quote in percent of nominal, as the exchange published it.""" + accrued_interest: Mapped[Decimal | None] + + +class PriceLast(Base): + """Latest known price per instrument — what the positions screen shows between refreshes.""" + + __tablename__ = "price_last" + + instrument_id: Mapped[int] = mapped_column( + ForeignKey("instrument.id", ondelete="CASCADE"), primary_key=True + ) + ts: Mapped[datetime] + price: Mapped[Decimal] + currency: Mapped[str] = mapped_column(String(3)) + source: Mapped[str] = mapped_column(String(16)) + + +class PriceManual(Base): + """User-set prices for what no exchange quotes: real estate, crypto, custom holdings.""" + + __tablename__ = "price_manual" + + id: Mapped[int] = mapped_column(primary_key=True) + instrument_id: Mapped[int] = mapped_column(ForeignKey("instrument.id", ondelete="CASCADE")) + d: Mapped[date] + price: Mapped[Decimal] + currency: Mapped[str] = mapped_column(String(3)) + note: Mapped[str | None] = mapped_column(String(256)) + + +class CorporateActionKind(enum.StrEnum): + dividend = "dividend" + coupon = "coupon" + amortization = "amortization" + repayment = "repayment" + stock_split = "split" + offer = "offer" + + +class CorporateActionStatus(enum.StrEnum): + forecast = "forecast" + """Projected from history — never mixed with announced money in the paid totals.""" + announced = "announced" + paid = "paid" + cancelled = "cancelled" + + +class CorporateAction(TimestampMixin, Base): + """Declared and projected payouts, the source of the dividend/coupon calendar.""" + + __tablename__ = "corporate_action" + __table_args__ = ( + UniqueConstraint("instrument_id", "kind", "source", "source_id"), + Index("ix_corporate_action_pay_date", "pay_date"), + ) + + id: Mapped[int] = mapped_column(primary_key=True) + instrument_id: Mapped[int] = mapped_column(ForeignKey("instrument.id", ondelete="CASCADE")) + kind: Mapped[CorporateActionKind] = mapped_column( + db_enum(CorporateActionKind, "corporate_action_kind") + ) + status: Mapped[CorporateActionStatus] = mapped_column( + db_enum(CorporateActionStatus, "corporate_action_status") + ) + record_date: Mapped[date | None] + ex_date: Mapped[date | None] + pay_date: Mapped[date | None] + amount_per_unit: Mapped[Decimal | None] + currency: Mapped[str | None] = mapped_column(String(3)) + ratio: Mapped[Decimal | None] + """Split ratio; NULL for cash events.""" + source: Mapped[str] = mapped_column(String(16)) + source_id: Mapped[str | None] = mapped_column(String(128)) + + +class PositionSnapshot(Base): + """What the broker says a position is — kept for reconciliation, never for analytics.""" + + __tablename__ = "position_snapshot" + + account_id: Mapped[int] = mapped_column( + ForeignKey("account.id", ondelete="CASCADE"), primary_key=True + ) + instrument_id: Mapped[int] = mapped_column( + ForeignKey("instrument.id", ondelete="CASCADE"), primary_key=True + ) + as_of: Mapped[datetime] = mapped_column(primary_key=True) + source: Mapped[str] = mapped_column(String(16), primary_key=True) + qty: Mapped[Decimal] + avg_price: Mapped[Decimal | None] + market_value: Mapped[Decimal | None] + currency: Mapped[str | None] = mapped_column(String(3)) + + +class CashSnapshot(Base): + """Broker-reported cash per currency — the counterpart of `position_snapshot`.""" + + __tablename__ = "cash_snapshot" + + account_id: Mapped[int] = mapped_column( + ForeignKey("account.id", ondelete="CASCADE"), primary_key=True + ) + currency: Mapped[str] = mapped_column(String(3), primary_key=True) + as_of: Mapped[datetime] = mapped_column(primary_key=True) + source: Mapped[str] = mapped_column(String(16), primary_key=True) + balance: Mapped[Decimal] + blocked: Mapped[Decimal | None] diff --git a/backend/src/fintracker/models/zenmoney.py b/backend/src/fintracker/models/zenmoney.py new file mode 100644 index 0000000..9950710 --- /dev/null +++ b/backend/src/fintracker/models/zenmoney.py @@ -0,0 +1,204 @@ +"""ZenMoney side: raw diff payloads, categories (tags), merchants, everyday transactions, +hand-written rules and trips (plan §1.7).""" + +from __future__ import annotations + +import enum +from datetime import date, datetime +from decimal import Decimal +from typing import Any + +from sqlalchemy import ( + BigInteger, + Boolean, + ForeignKey, + Integer, + String, + Text, + UniqueConstraint, + func, +) +from sqlalchemy.orm import Mapped, mapped_column + +from fintracker.db.base import Base, TimestampMixin, db_enum + + +class RawZenmoneyEntity(Base): + """Untouched JSON of every entity the /v8/diff endpoint returned; upserted by (type, id).""" + + __tablename__ = "raw_zenmoney_entity" + + entity_type: Mapped[str] = mapped_column(String(32), primary_key=True) + id: Mapped[str] = mapped_column(String(64), primary_key=True) + changed: Mapped[int | None] = mapped_column(BigInteger) + payload: Mapped[dict[str, Any]] + ingested_at: Mapped[datetime] = mapped_column(server_default=func.now()) + + +class RawZenmoneyDeletion(Base): + """Deletions reported by diff; the raw row is removed, this keeps the fact for audit.""" + + __tablename__ = "raw_zenmoney_deletion" + + entity_type: Mapped[str] = mapped_column(String(32), primary_key=True) + id: Mapped[str] = mapped_column(String(64), primary_key=True) + stamp: Mapped[int | None] = mapped_column(BigInteger) + deleted_at: Mapped[datetime] = mapped_column(server_default=func.now()) + + +class Category(TimestampMixin, Base): + """ZenMoney tag tree (max one level of nesting on their side).""" + + __tablename__ = "category" + __table_args__ = (UniqueConstraint("source", "source_id"),) + + id: Mapped[int] = mapped_column(primary_key=True) + source: Mapped[str] = mapped_column(String(32), default="zenmoney") + source_id: Mapped[str] = mapped_column(String(64)) + parent_id: Mapped[int | None] = mapped_column(ForeignKey("category.id", ondelete="SET NULL")) + name: Mapped[str] = mapped_column(String(256)) + icon: Mapped[str | None] = mapped_column(String(64)) + color: Mapped[int | None] = mapped_column(BigInteger) + show_income: Mapped[bool] = mapped_column(Boolean, default=True) + show_outcome: Mapped[bool] = mapped_column(Boolean, default=True) + archived: Mapped[bool] = mapped_column(Boolean, default=False) + + +class Merchant(Base): + __tablename__ = "merchant" + __table_args__ = (UniqueConstraint("source", "source_id"),) + + id: Mapped[int] = mapped_column(primary_key=True) + source: Mapped[str] = mapped_column(String(32), default="zenmoney") + source_id: Mapped[str] = mapped_column(String(64)) + name: Mapped[str] = mapped_column(String(256)) + + +class FlowType(enum.StrEnum): + income = "income" + expense = "expense" + internal_transfer = "internal_transfer" + savings_transfer = "savings_transfer" + broker_external_flow = "broker_external_flow" + deleted = "deleted" + other = "other" + + +class CashTxn(TimestampMixin, Base): + """One ZenMoney transaction. A transfer between own accounts is ONE row with both + income and outcome > 0. Native amounts only — conversion happens in analytics.""" + + __tablename__ = "cash_txn" + __table_args__ = (UniqueConstraint("source", "source_id"),) + + id: Mapped[int] = mapped_column(primary_key=True) + source: Mapped[str] = mapped_column(String(32), default="zenmoney") + source_id: Mapped[str] = mapped_column(String(64)) + ts: Mapped[datetime] + """created timestamp from ZenMoney (UTC).""" + date: Mapped[date] = mapped_column(index=True) + income: Mapped[Decimal] = mapped_column(default=Decimal(0)) + income_currency: Mapped[str | None] = mapped_column(String(3)) + income_account_id: Mapped[int | None] = mapped_column( + ForeignKey("account.id", ondelete="SET NULL"), index=True + ) + outcome: Mapped[Decimal] = mapped_column(default=Decimal(0)) + outcome_currency: Mapped[str | None] = mapped_column(String(3)) + outcome_account_id: Mapped[int | None] = mapped_column( + ForeignKey("account.id", ondelete="SET NULL"), index=True + ) + op_income: Mapped[Decimal | None] + op_income_currency: Mapped[str | None] = mapped_column(String(3)) + op_outcome: Mapped[Decimal | None] + op_outcome_currency: Mapped[str | None] = mapped_column(String(3)) + payee: Mapped[str | None] = mapped_column(String(512)) + original_payee: Mapped[str | None] = mapped_column(String(512)) + merchant_id: Mapped[int | None] = mapped_column(ForeignKey("merchant.id", ondelete="SET NULL")) + comment: Mapped[str | None] = mapped_column(Text) + mcc: Mapped[int | None] = mapped_column(Integer) + hold: Mapped[bool] = mapped_column(Boolean, default=False) + deleted: Mapped[bool] = mapped_column(Boolean, default=False) + changed: Mapped[int | None] = mapped_column(BigInteger) + primary_category_id: Mapped[int | None] = mapped_column( + ForeignKey("category.id", ondelete="SET NULL") + ) + """First tag as ZenMoney sent it (raw category).""" + + # --- derived by analytics/classify (rules applied); rebuilt on every refresh --- + flow_type: Mapped[FlowType] = mapped_column( + db_enum(FlowType, "flow_type"), default=FlowType.other + ) + category_id: Mapped[int | None] = mapped_column(ForeignKey("category.id", ondelete="SET NULL")) + """Effective category after rules (defaults to primary_category_id).""" + payee_canonical: Mapped[str | None] = mapped_column(String(512)) + is_one_off: Mapped[bool] = mapped_column(Boolean, default=False) + trip_id: Mapped[int | None] = mapped_column(ForeignKey("trip.id", ondelete="SET NULL")) + meta: Mapped[dict[str, Any] | None] + + +class CashTxnTag(Base): + """All tags of a transaction in ZenMoney order (ord 0 == primary).""" + + __tablename__ = "cash_txn_tag" + + txn_id: Mapped[int] = mapped_column( + ForeignKey("cash_txn.id", ondelete="CASCADE"), primary_key=True + ) + ord: Mapped[int] = mapped_column(Integer, primary_key=True) + category_id: Mapped[int] = mapped_column( + ForeignKey("category.id", ondelete="CASCADE"), index=True + ) + + +class RuleKind(enum.StrEnum): + savings = "savings" + """expense -> savings_transfer (money moved into savings, not consumed)""" + one_off = "one_off" + """still an expense, but out of the baseline and runway divisor""" + category = "category" + """override category; `value` = category name""" + payee = "payee" + """canonical payee name; `value` = the name""" + broker_target = "broker_target" + """outflow funds a brokerage account; `value` = broker account id (str)""" + ignore = "ignore" + """exclude from every metric: `flow_type` becomes `other`, and rule application STOPS for + that transaction — no later rule (by priority) can put it back into a counted flow""" + + +class RuleMatchType(enum.StrEnum): + id = "id" + payee = "payee" + comment = "comment" + category = "category" + mcc = "mcc" + account = "account" + + +class Rule(TimestampMixin, Base): + """The judgement layer: ZenMoney knows what an operation *was*, rules say what it *meant*. + `pattern` is matched case-insensitively with SQL LIKE semantics (`%` wildcards) except for + `id`, `mcc` and `account`, which compare exactly.""" + + __tablename__ = "rule" + + id: Mapped[int] = mapped_column(primary_key=True) + kind: Mapped[RuleKind] = mapped_column(db_enum(RuleKind, "rule_kind")) + match_type: Mapped[RuleMatchType] = mapped_column(db_enum(RuleMatchType, "rule_match_type")) + pattern: Mapped[str] = mapped_column(String(512)) + value: Mapped[str | None] = mapped_column(String(512)) + note: Mapped[str | None] = mapped_column(Text) + enabled: Mapped[bool] = mapped_column(Boolean, default=True) + priority: Mapped[int] = mapped_column(Integer, default=100) + last_matched_at: Mapped[datetime | None] + match_count: Mapped[int] = mapped_column(Integer, default=0) + + +class Trip(TimestampMixin, Base): + __tablename__ = "trip" + + id: Mapped[int] = mapped_column(primary_key=True) + name: Mapped[str] = mapped_column(String(256)) + date_from: Mapped[date] + date_to: Mapped[date] + country: Mapped[str | None] = mapped_column(String(2)) diff --git a/backend/src/fintracker/sources/__init__.py b/backend/src/fintracker/sources/__init__.py new file mode 100644 index 0000000..6eaa807 --- /dev/null +++ b/backend/src/fintracker/sources/__init__.py @@ -0,0 +1,13 @@ +"""Data sources. + +Importing this package pulls in every source package, and each of those calls +`registry.register(...)` at import time — so the API, the worker and the CLI all see the +same set of sources just by importing `fintracker.sources`. + +`sources/base.py` and `sources/registry.py` must never import the source packages back: +that is what keeps this import graph acyclic. +""" + +from fintracker.sources import cbr, moex, tinvest, zenmoney + +__all__ = ["cbr", "moex", "tinvest", "zenmoney"] diff --git a/backend/src/fintracker/sources/base.py b/backend/src/fintracker/sources/base.py new file mode 100644 index 0000000..206896a --- /dev/null +++ b/backend/src/fintracker/sources/base.py @@ -0,0 +1,39 @@ +"""Contract every data source implements (plan §2.3). + +A source pulls from one upstream (ZenMoney, T-Invest, MOEX, CBR) into the raw tier and +optionally maps into core tables. It owns its cursor via `SyncContext` and reports what +it did in `SyncResult`; the worker handles locking, run logging and scheduling. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Protocol + +from sqlalchemy.ext.asyncio import AsyncSession + +from fintracker.config import Settings + + +@dataclass +class SyncResult: + cursor_after: str | None = None + counts: dict[str, int] = field(default_factory=dict) + warnings: list[str] = field(default_factory=list) + changed: bool = True + """False when nothing new landed — lets the worker skip the metrics refresh.""" + + +@dataclass +class SyncContext: + session: AsyncSession + settings: Settings + cursor_before: str | None + triggered_by: str + extra: dict[str, Any] = field(default_factory=dict) + + +class Source(Protocol): + name: str + + async def sync(self, ctx: SyncContext) -> SyncResult: ... diff --git a/backend/src/fintracker/sources/cbr/__init__.py b/backend/src/fintracker/sources/cbr/__init__.py new file mode 100644 index 0000000..3541efa --- /dev/null +++ b/backend/src/fintracker/sources/cbr/__init__.py @@ -0,0 +1,8 @@ +"""CBR FX source package. Importing it registers the source.""" + +from fintracker.sources.cbr.sync import CbrSource +from fintracker.sources.registry import register + +register(CbrSource()) + +__all__ = ["CbrSource"] diff --git a/backend/src/fintracker/sources/cbr/client.py b/backend/src/fintracker/sources/cbr/client.py new file mode 100644 index 0000000..6fe2f95 --- /dev/null +++ b/backend/src/fintracker/sources/cbr/client.py @@ -0,0 +1,163 @@ +"""CBR (Bank of Russia) official FX rates over the public XML scripts. + +Two endpoints: + +* `GET /scripts/XML_daily.asp?date_req=DD/MM/YYYY` — every quoted currency on one day; + used only for the `CharCode -> internal id` map ('USD' -> 'R01235'). +* `GET /scripts/XML_dynamic.asp?date_req1=…&date_req2=…&VAL_NM_RQ=R01235` — one currency + over a date range, `192,1234`. + +Both answer windows-1251 XML with a decimal comma, and quote RUB per `Nominal` units +(100 for JPY, 10 for CNY at times). Values are parsed into `Decimal` and kept exactly as +printed, nominal included — dividing per unit happens later in `pricing/fx.py`. + +NETWORK NOTE: `trust_env=False`. This host exports http_proxy/https_proxy/all_proxy, and +cbr.ru fails the TLS handshake through that proxy, so the CBR client must bypass the +environment entirely. (ZenMoney is the opposite case and keeps the proxy — see +`sources/zenmoney/client.py`.) +""" + +from __future__ import annotations + +import xml.etree.ElementTree as ET +from dataclasses import dataclass +from datetime import date, datetime, timedelta +from decimal import Decimal, InvalidOperation + +import httpx + +DAILY_URL = "https://www.cbr.ru/scripts/XML_daily.asp" +DYNAMIC_URL = "https://www.cbr.ru/scripts/XML_dynamic.asp" +ENCODING = "windows-1251" +TIMEOUT = 60.0 +MAX_RANGE_DAYS = 366 +"""CBR copes with longer ranges, but requests are chunked to at most a year to be polite.""" + + +class CbrError(RuntimeError): + """CBR refused or returned something unparsable.""" + + +@dataclass(frozen=True) +class CbrQuote: + rate_date: date + nominal: int + value: Decimal + """RUB per `nominal` units, exactly as CBR printed it.""" + + +def new_http_client() -> httpx.AsyncClient: + # trust_env=False: see NETWORK NOTE above + return httpx.AsyncClient(trust_env=False, timeout=TIMEOUT) + + +def _fmt(day: date) -> str: + return day.strftime("%d/%m/%Y") + + +def _decimal(text: str | None) -> Decimal | None: + if not text: + return None + try: + return Decimal(text.replace("\xa0", "").replace(" ", "").replace(",", ".")) + except InvalidOperation: + return None + + +def _parse_xml(payload: bytes) -> ET.Element: + try: + return ET.fromstring(payload.decode(ENCODING)) + except (ET.ParseError, UnicodeDecodeError) as exc: + raise CbrError(f"CBR returned unparsable XML: {exc}") from exc + + +def parse_code_map(payload: bytes) -> dict[str, str]: + """XML_daily -> {'USD': 'R01235', …}.""" + root = _parse_xml(payload) + codes: dict[str, str] = {} + for node in root.iter("Valute"): + char_code = (node.findtext("CharCode") or "").strip().upper() + cbr_id = (node.get("ID") or "").strip() + if char_code and cbr_id: + codes[char_code] = cbr_id + return codes + + +def parse_dynamic(payload: bytes) -> list[CbrQuote]: + """XML_dynamic -> quotes, nominal and value untouched.""" + root = _parse_xml(payload) + quotes: list[CbrQuote] = [] + for node in root.iter("Record"): + raw_date = node.get("Date") + value = _decimal(node.findtext("Value")) + if not raw_date or value is None: + continue + try: + rate_date = datetime.strptime(raw_date, "%d.%m.%Y").date() + except ValueError: + continue + nominal_text = (node.findtext("Nominal") or "1").strip() + try: + nominal = int(nominal_text.replace("\xa0", "").replace(" ", "")) or 1 + except ValueError: + nominal = 1 + quotes.append(CbrQuote(rate_date=rate_date, nominal=nominal, value=value)) + return quotes + + +def split_range(start: date, end: date, max_days: int = MAX_RANGE_DAYS) -> list[tuple[date, date]]: + """Chunk a long range into windows of at most `max_days` days.""" + if start > end: + return [] + windows: list[tuple[date, date]] = [] + window_start = start + while window_start <= end: + window_end = min(end, window_start + timedelta(days=max_days - 1)) + windows.append((window_start, window_end)) + window_start = window_end + timedelta(days=1) + return windows + + +class CbrClient: + def __init__(self, http: httpx.AsyncClient | None = None) -> None: + self._http = http + self._owns_http = http is None + + def _client(self) -> httpx.AsyncClient: + if self._http is None: + self._http = new_http_client() + return self._http + + async def aclose(self) -> None: + if self._http is not None and self._owns_http: + await self._http.aclose() + self._http = None + + async def __aenter__(self) -> CbrClient: + return self + + async def __aexit__(self, *exc_info: object) -> None: + await self.aclose() + + async def _get(self, url: str, params: dict[str, str]) -> bytes: + resp = await self._client().get(url, params=params) + if resp.status_code >= 400: + raise CbrError(f"{url} failed: HTTP {resp.status_code}") + return resp.content + + async def code_map(self, on: date) -> dict[str, str]: + return parse_code_map(await self._get(DAILY_URL, {"date_req": _fmt(on)})) + + async def quotes(self, cbr_id: str, start: date, end: date) -> list[CbrQuote]: + collected: list[CbrQuote] = [] + for window_start, window_end in split_range(start, end): + payload = await self._get( + DYNAMIC_URL, + { + "date_req1": _fmt(window_start), + "date_req2": _fmt(window_end), + "VAL_NM_RQ": cbr_id, + }, + ) + collected.extend(parse_dynamic(payload)) + return collected diff --git a/backend/src/fintracker/sources/cbr/sync.py b/backend/src/fintracker/sources/cbr/sync.py new file mode 100644 index 0000000..c1b7a9c --- /dev/null +++ b/backend/src/fintracker/sources/cbr/sync.py @@ -0,0 +1,161 @@ +"""The `cbr` source: official RUB rates for every currency the ledger actually uses. + +What to fetch is derived from the data, not configured: the distinct currencies of +`account.currency`, `cash_txn.income_currency` and `cash_txn.outcome_currency` (RUB itself +is quoted as 1.0 by `pricing/fx.py`, so it is never requested). + +Date range: from the earliest transaction minus 7 days (or today − 30 days when there are +no transactions yet) up to today — `analytics.today_local()`, the deployment timezone, the +same "today" the metrics end on. MSK runs ahead of UTC, so a UTC date would ask for one day +less for the first hours of every Moscow morning. Incrementally, the cursor is the last date +fetched and the next run starts 3 days earlier — CBR publishes the next business day's rate +in the evening, and that overlap re-reads the tail cheaply. If the ledger grew *backwards* (an +import of older history) the full range is used again, detected by comparing the wanted +start with the earliest date already in `raw_cbr_rate`. + +Currencies CBR does not quote (crypto, metals such as XAU) are reported in `warnings` and +skipped — never an error, and never an invented rate. +""" + +from __future__ import annotations + +import logging +from datetime import UTC, date, datetime, timedelta + +from sqlalchemy import func, select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from fintracker.analytics import ledger_date_range, today_local +from fintracker.models import Account, CashTxn, RawCbrRate +from fintracker.sources.base import SyncContext, SyncResult +from fintracker.sources.cbr.client import CbrClient, CbrQuote + +log = logging.getLogger(__name__) + +SOURCE = "cbr" +BASE = "RUB" +LOOKBACK_DAYS = 7 +"""Margin before the first transaction, so a rate exists for its date even over a weekend.""" +CURSOR_OVERLAP_DAYS = 3 +BACKFILL_TOLERANCE_DAYS = 14 +"""How far the first stored quote may legitimately sit after the wanted start (holidays).""" +NO_HISTORY_DAYS = 30 +CHUNK = 500 + + +class CbrSource: + name = SOURCE + + async def sync(self, ctx: SyncContext) -> SyncResult: + session = ctx.session + today = today_local() + currencies = await currencies_in_use(session) + if not currencies: + log.info("cbr: only %s in use, nothing to fetch", BASE) + return SyncResult( + cursor_after=today.isoformat(), + counts={"currencies": 0, "rates": 0}, + changed=False, + ) + + start = await _start_date(session, ctx.cursor_before, today) + warnings: list[str] = [] + stored = 0 + async with CbrClient() as client: + codes = await client.code_map(today) + fetched_currencies = 0 + for ccy in currencies: + cbr_id = codes.get(ccy) + if not cbr_id: + warnings.append(f"{ccy} is not quoted by CBR — skipped (crypto or metal?)") + continue + quotes = await client.quotes(cbr_id, start, today) + stored += await _store(session, ccy, quotes) + fetched_currencies += 1 + + await session.commit() + log.info( + "cbr: %s rates for %s currencies, %s..%s%s", + stored, + fetched_currencies, + start, + today, + f", skipped {len(warnings)}" if warnings else "", + ) + return SyncResult( + cursor_after=today.isoformat(), + counts={"currencies": fetched_currencies, "rates": stored}, + warnings=warnings, + changed=stored > 0, + ) + + +async def currencies_in_use(session: AsyncSession) -> list[str]: + found: set[str] = set() + rows = await session.execute(select(Account.currency).where(Account.currency.is_not(None))) + found.update(code for code in rows.scalars().all() if code) + # deleted transactions are tombstones no metric reads — their currency is not "in use" + for column in (CashTxn.income_currency, CashTxn.outcome_currency): + rows = await session.execute( + select(column).where(column.is_not(None), CashTxn.deleted.is_(False)).distinct() + ) + found.update(code for code in rows.scalars().all() if code) + return sorted(code for code in found if code != BASE) + + +async def _start_date(session: AsyncSession, cursor: str | None, today: date) -> date: + first_txn, _ = await ledger_date_range(session) + full_start = ( + first_txn - timedelta(days=LOOKBACK_DAYS) + if first_txn is not None + else today - timedelta(days=NO_HISTORY_DAYS) + ) + cursor_date = _parse_date(cursor) + if cursor_date is None: + return full_start + have_from = (await session.execute(select(func.min(RawCbrRate.rate_date)))).scalar_one_or_none() + if have_from is None or full_start + timedelta(days=BACKFILL_TOLERANCE_DAYS) < have_from: + # the ledger grew backwards (older history imported) — refetch from the new beginning. + # The tolerance absorbs the normal case where the wanted start simply falls on a + # weekend or a holiday stretch that CBR never quoted. + return full_start + return min(today, cursor_date - timedelta(days=CURSOR_OVERLAP_DAYS)) + + +def _parse_date(value: str | None) -> date | None: + if not value: + return None + try: + return date.fromisoformat(value) + except ValueError: + log.warning("cbr: unusable cursor %r, refetching the full range", value) + return None + + +async def _store(session: AsyncSession, ccy: str, quotes: list[CbrQuote]) -> int: + rows = [ + { + "rate_date": quote.rate_date, + "ccy": ccy, + "nominal": quote.nominal, + "value": quote.value, + "fetched_at": datetime.now(UTC), + } + for quote in quotes + ] + if not rows: + return 0 + for start in range(0, len(rows), CHUNK): + chunk = rows[start : start + CHUNK] + stmt = pg_insert(RawCbrRate).values(chunk) + stmt = stmt.on_conflict_do_update( + index_elements=["rate_date", "ccy"], + set_={ + "nominal": stmt.excluded.nominal, + "value": stmt.excluded.value, + "fetched_at": stmt.excluded.fetched_at, + }, + ) + await session.execute(stmt) + return len(rows) diff --git a/backend/src/fintracker/sources/registry.py b/backend/src/fintracker/sources/registry.py new file mode 100644 index 0000000..211e85a --- /dev/null +++ b/backend/src/fintracker/sources/registry.py @@ -0,0 +1,24 @@ +"""Registry of available sources. Later phases register zenmoney, cbr, tinvest, moex here.""" + +from __future__ import annotations + +from fintracker.sources.base import Source + +_SOURCES: dict[str, Source] = {} + + +def register(source: Source) -> Source: + _SOURCES[source.name] = source + return source + + +def unregister(name: str) -> None: + _SOURCES.pop(name, None) + + +def get(name: str) -> Source: + return _SOURCES[name] + + +def names() -> list[str]: + return sorted(_SOURCES) diff --git a/backend/src/fintracker/sources/zenmoney/__init__.py b/backend/src/fintracker/sources/zenmoney/__init__.py new file mode 100644 index 0000000..f525363 --- /dev/null +++ b/backend/src/fintracker/sources/zenmoney/__init__.py @@ -0,0 +1,8 @@ +"""ZenMoney source package. Importing it registers the source.""" + +from fintracker.sources.registry import register +from fintracker.sources.zenmoney.sync import ZenmoneySource + +register(ZenmoneySource()) + +__all__ = ["ZenmoneySource"] diff --git a/backend/src/fintracker/sources/zenmoney/client.py b/backend/src/fintracker/sources/zenmoney/client.py new file mode 100644 index 0000000..524a802 --- /dev/null +++ b/backend/src/fintracker/sources/zenmoney/client.py @@ -0,0 +1,280 @@ +"""HTTP access to the ZenMoney API (plan § "ZenMoney"). + +Reading is possible only through one endpoint: `POST https://api.zenmoney.ru/v8/diff/`. +It is incremental by `serverTimestamp`; the first call (cursor 0) additionally asks for +`forceFetch` of every entity type. The answer holds the same entity arrays (only rows +changed since the cursor), a `deletion` list and the new `serverTimestamp`. + +NETWORK NOTE: this host exports http_proxy/https_proxy/all_proxy and api.zenmoney.ru IS +reachable through that proxy, so the client keeps httpx's default `trust_env=True`. +The CBR client does the opposite — cbr.ru fails the TLS handshake through the proxy, +see `sources/cbr/client.py`. + +AUTH. Access tokens live 86400 s. Two modes, picked from settings: + +* static token — `ZENMONEY_TOKEN` (e.g. minted at zerro.app/token). There is nothing to + rotate, so HTTP 401 can only be reported as "renew ZENMONEY_TOKEN". +* OAuth rotation — enabled when `ZENMONEY_CLIENT_ID` and `ZENMONEY_CLIENT_SECRET` are set. + The `{access_token, refresh_token, expires_at}` triple lives in `source_credential` + (source='zenmoney'), seeded from `ZENMONEY_REFRESH_TOKEN`. Before a sync an expired + (or missing) access token is refreshed with + `POST https://api.zenmoney.ru/oauth2/token/` form-encoded + `grant_type=refresh_token&refresh_token=…&client_id=…&client_secret=…`; the response + `{access_token, refresh_token, expires_in, token_type}` replaces the stored pair. + A 401 on the diff call triggers exactly one extra refresh + retry. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from types import TracebackType +from typing import Any, Self + +import httpx +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from fintracker.config import Settings +from fintracker.models import SourceCredential + +log = logging.getLogger(__name__) + +SOURCE = "zenmoney" +DIFF_URL = "https://api.zenmoney.ru/v8/diff/" +TOKEN_URL = "https://api.zenmoney.ru/oauth2/token/" +TIMEOUT = 120.0 + +ENTITY_TYPES: tuple[str, ...] = ( + "instrument", + "company", + "user", + "account", + "tag", + "merchant", + "budget", + "reminder", + "reminderMarker", + "transaction", +) +"""Every entity type /v8/diff/ can return; also the `forceFetch` list on a full pull.""" + +EXPIRY_SKEW = timedelta(minutes=5) +DEFAULT_TOKEN_TTL = 86400 + + +class ZenmoneyError(RuntimeError): + """Upstream refused or misbehaved.""" + + +class ZenmoneyAuthError(ZenmoneyError): + """Credentials are missing, expired or rejected.""" + + +@dataclass +class TokenBundle: + access_token: str + refresh_token: str | None = None + expires_at: datetime | None = None + + @property + def expired(self) -> bool: + if self.expires_at is None: + return False + return self.expires_at - EXPIRY_SKEW <= datetime.now(UTC) + + def to_payload(self) -> dict[str, Any]: + return { + "access_token": self.access_token, + "refresh_token": self.refresh_token, + "expires_at": self.expires_at.isoformat() if self.expires_at else None, + } + + @classmethod + def from_payload(cls, payload: dict[str, Any] | None) -> TokenBundle | None: + if not payload or not payload.get("access_token"): + return None + raw_expires = payload.get("expires_at") + expires_at: datetime | None = None + if isinstance(raw_expires, str): + try: + expires_at = datetime.fromisoformat(raw_expires) + except ValueError: + expires_at = None + if expires_at is not None and expires_at.tzinfo is None: + expires_at = expires_at.replace(tzinfo=UTC) + return cls( + access_token=str(payload["access_token"]), + refresh_token=payload.get("refresh_token"), + expires_at=expires_at, + ) + + +class ZenmoneyClient: + """One diff call per sync, plus the token rotation around it. + + Needs the sync session because in OAuth mode the rotated pair must be persisted in + `source_credential` — the worker runs unattended, so a token it fetched and lost + would strand the next run. + """ + + def __init__( + self, + settings: Settings, + session: AsyncSession, + *, + http: httpx.AsyncClient | None = None, + ) -> None: + self._settings = settings + self._session = session + self._http = http + self._owns_http = http is None + self._token: TokenBundle | None = None + + # --- lifecycle --------------------------------------------------------------- + + @property + def oauth_mode(self) -> bool: + return bool(self._settings.zenmoney_client_id and self._settings.zenmoney_client_secret) + + def _client(self) -> httpx.AsyncClient: + if self._http is None: + # trust_env stays on: ZenMoney is reachable only through the host proxy + self._http = httpx.AsyncClient(timeout=TIMEOUT) + return self._http + + async def aclose(self) -> None: + if self._http is not None and self._owns_http: + await self._http.aclose() + self._http = None + + async def __aenter__(self) -> Self: + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + await self.aclose() + + # --- diff -------------------------------------------------------------------- + + async def diff(self, server_timestamp: int) -> dict[str, Any]: + body: dict[str, Any] = { + "currentClientTimestamp": int(datetime.now(UTC).timestamp()), + "serverTimestamp": server_timestamp, + } + if server_timestamp == 0: + body["forceFetch"] = list(ENTITY_TYPES) + + token = await self.access_token() + resp = await self._post_diff(token, body) + if resp.status_code == 401: + if not self.oauth_mode: + raise ZenmoneyAuthError( + "ZenMoney rejected the static access token (HTTP 401). ZenMoney access " + "tokens live 24 hours — renew ZENMONEY_TOKEN (e.g. at zerro.app/token) " + "and put the new value in .env, or configure ZENMONEY_CLIENT_ID / " + "ZENMONEY_CLIENT_SECRET / ZENMONEY_REFRESH_TOKEN for automatic rotation." + ) + log.info("zenmoney: diff got 401, refreshing the access token once") + refreshed = await self._refresh_token() + resp = await self._post_diff(refreshed.access_token, body) + if resp.status_code == 401: + raise ZenmoneyAuthError( + "ZenMoney rejected a freshly refreshed access token (HTTP 401); check " + "ZENMONEY_CLIENT_ID / ZENMONEY_CLIENT_SECRET and the refresh token " + "stored in source_credential." + ) + if resp.status_code >= 400: + raise ZenmoneyError( + f"ZenMoney /v8/diff/ failed: HTTP {resp.status_code} {resp.text[:500]}" + ) + data = json.loads(resp.text) + if not isinstance(data, dict): + raise ZenmoneyError("ZenMoney /v8/diff/ returned a non-object body") + return data + + async def _post_diff(self, token: str, body: dict[str, Any]) -> httpx.Response: + return await self._client().post( + DIFF_URL, + headers={"Authorization": f"Bearer {token}"}, + json=body, + ) + + # --- tokens ------------------------------------------------------------------ + + async def access_token(self) -> str: + if not self.oauth_mode: + if not self._settings.zenmoney_token: + raise ZenmoneyAuthError( + "ZENMONEY_TOKEN is not set — put a ZenMoney access token in .env " + "(e.g. from zerro.app/token), or configure ZENMONEY_CLIENT_ID / " + "ZENMONEY_CLIENT_SECRET / ZENMONEY_REFRESH_TOKEN for OAuth rotation." + ) + return self._settings.zenmoney_token + + bundle = self._token or await self._load_credential() + if bundle is None or bundle.expired: + bundle = await self._refresh_token(bundle) + self._token = bundle + return bundle.access_token + + async def _load_credential(self) -> TokenBundle | None: + row = await self._session.get(SourceCredential, SOURCE) + return TokenBundle.from_payload(row.payload if row else None) + + async def _refresh_token(self, bundle: TokenBundle | None = None) -> TokenBundle: + settings = self._settings + known = bundle or self._token or await self._load_credential() + refresh_token = (known.refresh_token if known else None) or settings.zenmoney_refresh_token + if not refresh_token: + raise ZenmoneyAuthError( + "OAuth mode is configured but no refresh token is available — set " + "ZENMONEY_REFRESH_TOKEN in .env to seed source_credential." + ) + resp = await self._client().post( + TOKEN_URL, + data={ + "grant_type": "refresh_token", + "refresh_token": refresh_token, + "client_id": settings.zenmoney_client_id or "", + "client_secret": settings.zenmoney_client_secret or "", + }, + ) + if resp.status_code >= 400: + raise ZenmoneyAuthError( + "ZenMoney refused to refresh the access token: " + f"HTTP {resp.status_code} {resp.text[:500]}" + ) + data = resp.json() + access_token = data.get("access_token") + if not access_token: + raise ZenmoneyAuthError("ZenMoney token response carried no access_token") + try: + ttl = int(data.get("expires_in") or DEFAULT_TOKEN_TTL) + except (TypeError, ValueError): + ttl = DEFAULT_TOKEN_TTL + fresh = TokenBundle( + access_token=str(access_token), + refresh_token=data.get("refresh_token") or refresh_token, + expires_at=datetime.now(UTC) + timedelta(seconds=ttl), + ) + await self._store_credential(fresh) + self._token = fresh + return fresh + + async def _store_credential(self, bundle: TokenBundle) -> None: + payload = bundle.to_payload() + stmt = pg_insert(SourceCredential).values(source=SOURCE, payload=payload) + stmt = stmt.on_conflict_do_update( + index_elements=["source"], + set_={"payload": stmt.excluded.payload, "updated_at": datetime.now(UTC)}, + ) + await self._session.execute(stmt) + # committed right away: a token we fetched but lost would strand the next run + await self._session.commit() diff --git a/backend/src/fintracker/sources/zenmoney/mapper.py b/backend/src/fintracker/sources/zenmoney/mapper.py new file mode 100644 index 0000000..4d55843 --- /dev/null +++ b/backend/src/fintracker/sources/zenmoney/mapper.py @@ -0,0 +1,483 @@ +"""Raw ZenMoney entities -> core tables (plan §1.2, §1.7). + +The mapper never looks at the diff itself: it rebuilds the core rows from the full +`raw_zenmoney_entity` table, so it is idempotent and self-healing (a mapping bug is fixed +by re-running it, no re-download). Personal volumes are ~10^4 transactions, so a full +upsert per sync is cheap; `sync.py` skips the mapper entirely when the diff was empty. + +Rows deleted upstream are gone from the raw table, so they are never re-created here — +the flags `sync.py` sets while processing `deletion` survive. + +Money: every amount goes through `_money()` into `Decimal`. ZenMoney sends amounts as JSON +numbers, so they arrive from JSONB as Python floats; `Decimal(str(value))` round-trips the +printed (≤2 decimals) value exactly. Nothing float-valued is ever stored. + +What is deliberately NOT touched here (owned by `analytics/classify.py`): `payee_canonical`, +`is_one_off`, `trip_id`, and the refined `flow_type`/`category_id`. The mapper only lays +down the base values ZenMoney itself implies. + +Account fields the USER owns (`PATCH /accounts/{id}`): `name`, `role`, `include_in_net_worth` +— plus `mirror_of_account_id` and `primary_event_source`, which the mapper never produces at +all. They are seeded on INSERT and never overwritten on conflict, so a sync cannot undo an +edit. The single exception: an account the source marks `archive` is forced out of net worth, +the same way `sync.py` treats a deletion. +""" + +from __future__ import annotations + +from collections.abc import Callable, Iterable, Iterator, Sequence +from datetime import UTC, date, datetime +from decimal import Decimal, InvalidOperation +from typing import Any + +from sqlalchemy import case, delete, false, select, update +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from fintracker.config import Settings +from fintracker.models import ( + Account, + AccountKind, + AccountRole, + CashTxn, + CashTxnTag, + Category, + FlowType, + Merchant, + RawZenmoneyEntity, +) + +SOURCE = "zenmoney" +CHUNK = 500 + +KIND_BY_TYPE: dict[str, AccountKind] = { + "cash": AccountKind.zm_cash, + "ccard": AccountKind.zm_card, + "checking": AccountKind.zm_checking, + "deposit": AccountKind.zm_deposit, + "loan": AccountKind.zm_loan, + "emoney": AccountKind.zm_emoney, + "debt": AccountKind.zm_debt, +} +"""ZenMoney `account.type` -> `AccountKind`. Only `ccard` -> `zm_card` is not a literal +"zm_" + type: the enum in `models/accounts.py` spells the card kind without the extra c.""" + +DEBT_TYPES = frozenset({"loan", "debt"}) +DEPOSIT_FIELDS = ( + "capitalization", + "percent", + "startDate", + "endDateOffset", + "endDateOffsetInterval", + "payoffStep", + "payoffInterval", +) +META_FIELDS = ("reminderMarker", "latitude", "longitude") + + +def _money(value: Any) -> Decimal | None: + if value is None or isinstance(value, bool): + return None + if isinstance(value, Decimal): + return value + try: + return Decimal(str(value)) + except (InvalidOperation, ValueError): + return None + + +def _int(value: Any) -> int | None: + if value is None or isinstance(value, bool): + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _ccy(codes: dict[int, str], value: Any) -> str | None: + """Instrument id -> currency code. `None` for an absent or unknown instrument; id 0 is a + perfectly valid key and must not collapse into a sentinel.""" + ident = _int(value) + return codes.get(ident) if ident is not None else None + + +def _day(value: Any) -> date | None: + if not isinstance(value, str): + return None + try: + return date.fromisoformat(value[:10]) + except ValueError: + return None + + +def _chunks( + rows: Sequence[dict[str, Any]], size: int = CHUNK +) -> Iterator[Sequence[dict[str, Any]]]: + for start in range(0, len(rows), size): + yield rows[start : start + size] + + +async def _upsert( + session: AsyncSession, + model: type[Any], + rows: Sequence[dict[str, Any]], + conflict: list[str], + update_cols: Iterable[str], + set_extra: Callable[[Any], dict[str, Any]] | None = None, +) -> int: + """Postgres INSERT … ON CONFLICT DO UPDATE, chunked. `update_cols` deliberately omits + the columns other layers own, so re-mapping cannot clobber them. `set_extra` gets the + insert statement and returns SET entries that are not a plain copy from `excluded`.""" + cols = list(update_cols) + for chunk in _chunks(rows): + stmt = pg_insert(model).values(list(chunk)) + set_: dict[str, Any] = {name: getattr(stmt.excluded, name) for name in cols} + if set_extra is not None: + set_.update(set_extra(stmt)) + stmt = stmt.on_conflict_do_update(index_elements=conflict, set_=set_) + await session.execute(stmt) + return len(rows) + + +async def _source_ids(session: AsyncSession, model: type[Any]) -> dict[str, int]: + rows = await session.execute(select(model.source_id, model.id).where(model.source == SOURCE)) + return {source_id: row_id for source_id, row_id in rows.all()} + + +async def load_raw(session: AsyncSession) -> dict[str, list[dict[str, Any]]]: + """Every raw entity, grouped by type.""" + rows = await session.execute(select(RawZenmoneyEntity.entity_type, RawZenmoneyEntity.payload)) + grouped: dict[str, list[dict[str, Any]]] = {} + for entity_type, payload in rows.all(): + if isinstance(payload, dict): + grouped.setdefault(entity_type, []).append(payload) + return grouped + + +def instrument_codes(instruments: list[dict[str, Any]]) -> dict[int, str]: + """ZenMoney instrument id -> 3-letter code ('RUB').""" + codes: dict[int, str] = {} + for payload in instruments: + ident = _int(payload.get("id")) + code = payload.get("shortTitle") or payload.get("title") + if ident is not None and isinstance(code, str) and code: + codes[ident] = code.upper()[:3] + return codes + + +async def map_all(session: AsyncSession, settings: Settings) -> tuple[dict[str, int], list[str]]: + warnings: list[str] = [] + raw = await load_raw(session) + codes = instrument_codes(raw.get("instrument", [])) + + categories = await _map_categories(session, raw.get("tag", [])) + merchants = await _map_merchants(session, raw.get("merchant", [])) + accounts = await _map_accounts(session, raw.get("account", []), codes, settings, warnings) + transactions = await _map_transactions( + session, + raw.get("transaction", []), + codes=codes, + accounts=accounts, + categories=categories, + merchants=merchants, + warnings=warnings, + ) + counts = { + "instruments": len(codes), + "categories": len(categories), + "merchants": len(merchants), + "accounts": len(accounts), + "transactions": transactions, + } + return counts, warnings + + +# --- tags -> category ----------------------------------------------------------------- + + +async def _map_categories(session: AsyncSession, tags: list[dict[str, Any]]) -> dict[str, int]: + """Two passes: upsert the rows, then resolve `parent_id` (ZenMoney nests one level).""" + rows = [ + { + "source": SOURCE, + "source_id": str(payload["id"]), + "name": str(payload.get("title") or ""), + "icon": payload.get("icon"), + "color": _int(payload.get("color")), + "show_income": bool(payload.get("showIncome", True)), + "show_outcome": bool(payload.get("showOutcome", True)), + "archived": False, + } + for payload in tags + if payload.get("id") is not None + ] + if rows: + await _upsert( + session, + Category, + rows, + ["source", "source_id"], + ("name", "icon", "color", "show_income", "show_outcome"), + ) + ids = await _source_ids(session, Category) + if not ids: + return ids + + parents = [ + {"id": ids[str(payload["id"])], "parent_id": ids.get(str(payload.get("parent")))} + for payload in tags + if payload.get("id") is not None and str(payload["id"]) in ids + ] + if parents: + await session.execute(update(Category), parents) + return ids + + +async def _map_merchants(session: AsyncSession, merchants: list[dict[str, Any]]) -> dict[str, int]: + rows = [ + { + "source": SOURCE, + "source_id": str(payload["id"]), + "name": str(payload.get("title") or ""), + } + for payload in merchants + if payload.get("id") is not None + ] + if rows: + await _upsert(session, Merchant, rows, ["source", "source_id"], ("name",)) + return await _source_ids(session, Merchant) + + +# --- accounts ------------------------------------------------------------------------- + + +def account_role(acc_type: str, savings: bool) -> AccountRole: + """loan/debt are what we owe; an explicit savings flag or a deposit is savings money; + everything else is spendable. `investment` is left to broker accounts (phase 2), and + `mirror_of_account_id` / `primary_event_source` stay whatever the user set.""" + if acc_type in DEBT_TYPES: + return AccountRole.debt + if savings or acc_type == "deposit": + return AccountRole.savings + return AccountRole.liquid + + +async def _map_accounts( + session: AsyncSession, + accounts: list[dict[str, Any]], + codes: dict[int, str], + settings: Settings, + warnings: list[str], +) -> dict[str, int]: + observed_at = datetime.now(UTC) + rows: list[dict[str, Any]] = [] + for payload in accounts: + if payload.get("id") is None: + continue + source_id = str(payload["id"]) + acc_type = str(payload.get("type") or "cash") + kind = KIND_BY_TYPE.get(acc_type) + if kind is None: + warnings.append( + f"account {source_id}: unknown ZenMoney type {acc_type!r}, stored as zm_cash" + ) + kind = AccountKind.zm_cash + currency = _ccy(codes, payload.get("instrument")) + if not currency: + currency = settings.base_currency + warnings.append( + f"account {source_id}: instrument {payload.get('instrument')!r} is unknown, " + f"currency assumed {currency}" + ) + archived = bool(payload.get("archive")) + terms = { + field: payload[field] for field in DEPOSIT_FIELDS if payload.get(field) is not None + } + rows.append( + { + "source": SOURCE, + "source_id": source_id, + "kind": kind, + # name / role / include_in_net_worth are seeded here and then owned by the + # user: they are absent from the ON CONFLICT set below + "name": str(payload.get("title") or source_id), + "currency": currency, + "role": account_role(acc_type, bool(payload.get("savings"))), + # inBalance is ZenMoney's own "counts towards my balance" switch + "include_in_net_worth": bool(payload.get("inBalance", True)) and not archived, + "archived": archived, + "opened_at": _day(payload.get("startDate")), + "deposit_terms": terms or None, + "balance": _money(payload.get("balance")), + "start_balance": _money(payload.get("startBalance")), + "credit_limit": _money(payload.get("creditLimit")), + "balance_as_of": observed_at, + } + ) + if rows: + await _upsert( + session, + Account, + rows, + ["source", "source_id"], + ( + "kind", + "currency", + "archived", + "opened_at", + "deposit_terms", + "balance", + "start_balance", + "credit_limit", + "balance_as_of", + ), + # archived upstream still forces the account out of net worth (same as a + # deletion); otherwise the user's own switch wins + set_extra=lambda stmt: { + "include_in_net_worth": case( + (stmt.excluded.archived, false()), + else_=Account.include_in_net_worth, + ) + }, + ) + return await _source_ids(session, Account) + + +# --- transactions --------------------------------------------------------------------- + + +def flow_type(income: Decimal, outcome: Decimal, deleted: bool) -> FlowType: + """The base classification ZenMoney itself implies. Rules (savings_transfer, one_off, + broker flows) refine it later in `analytics/classify.py`.""" + if deleted: + return FlowType.deleted + if income > 0 and outcome > 0: + return FlowType.internal_transfer + if outcome > 0: + return FlowType.expense + if income > 0: + return FlowType.income + return FlowType.other + + +async def _map_transactions( + session: AsyncSession, + transactions: list[dict[str, Any]], + *, + codes: dict[int, str], + accounts: dict[str, int], + categories: dict[str, int], + merchants: dict[str, int], + warnings: list[str], +) -> int: + rows: list[dict[str, Any]] = [] + tags_by_source_id: dict[str, list[int]] = {} + for payload in transactions: + if payload.get("id") is None: + continue + source_id = str(payload["id"]) + day = _day(payload.get("date")) + if day is None: + warnings.append(f"transaction {source_id}: unparsable date {payload.get('date')!r}") + continue + created = _int(payload.get("created")) or _int(payload.get("changed")) + ts = ( + datetime.fromtimestamp(created, UTC) + if created is not None + else datetime.combine(day, datetime.min.time(), UTC) + ) + income = _money(payload.get("income")) or Decimal(0) + outcome = _money(payload.get("outcome")) or Decimal(0) + deleted = bool(payload.get("deleted")) + + tag_ids = [ + categories[str(tag)] for tag in (payload.get("tag") or []) if str(tag) in categories + ] + tags_by_source_id[source_id] = tag_ids + primary_category_id = tag_ids[0] if tag_ids else None + meta = {field: payload[field] for field in META_FIELDS if payload.get(field) is not None} + + rows.append( + { + "source": SOURCE, + "source_id": source_id, + "ts": ts, + "date": day, + "income": income, + "income_currency": _ccy(codes, payload.get("incomeInstrument")), + "income_account_id": accounts.get(str(payload.get("incomeAccount"))), + "outcome": outcome, + "outcome_currency": _ccy(codes, payload.get("outcomeInstrument")), + "outcome_account_id": accounts.get(str(payload.get("outcomeAccount"))), + "op_income": _money(payload.get("opIncome")), + "op_income_currency": _ccy(codes, payload.get("opIncomeInstrument")), + "op_outcome": _money(payload.get("opOutcome")), + "op_outcome_currency": _ccy(codes, payload.get("opOutcomeInstrument")), + "payee": payload.get("payee"), + "original_payee": payload.get("originalPayee"), + "merchant_id": merchants.get(str(payload.get("merchant"))), + "comment": payload.get("comment"), + "mcc": _int(payload.get("mcc")), + "hold": bool(payload.get("hold")), + "deleted": deleted, + "changed": _int(payload.get("changed")), + "primary_category_id": primary_category_id, + "flow_type": flow_type(income, outcome, deleted), + "category_id": primary_category_id, + "is_one_off": False, + } + | {"meta": meta or None} + ) + if not rows: + return 0 + + await _upsert( + session, + CashTxn, + rows, + ["source", "source_id"], + ( + "ts", + "date", + "income", + "income_currency", + "income_account_id", + "outcome", + "outcome_currency", + "outcome_account_id", + "op_income", + "op_income_currency", + "op_outcome", + "op_outcome_currency", + "payee", + "original_payee", + "merchant_id", + "comment", + "mcc", + "hold", + "deleted", + "changed", + "primary_category_id", + "flow_type", + "category_id", + "meta", + ), + ) + + txn_ids = await _source_ids(session, CashTxn) + links: list[dict[str, Any]] = [] + for source_id, tag_ids in tags_by_source_id.items(): + txn_id = txn_ids.get(source_id) + if txn_id is None: + continue + links.extend( + {"txn_id": txn_id, "ord": ord_, "category_id": category_id} + for ord_, category_id in enumerate(tag_ids) + ) + mapped_ids = [txn_ids[s] for s in tags_by_source_id if s in txn_ids] + if mapped_ids: + await session.execute(delete(CashTxnTag).where(CashTxnTag.txn_id.in_(mapped_ids))) + for chunk in _chunks(links): + await session.execute(pg_insert(CashTxnTag).values(list(chunk))) + return len(rows) diff --git a/backend/src/fintracker/sources/zenmoney/sync.py b/backend/src/fintracker/sources/zenmoney/sync.py new file mode 100644 index 0000000..36fdc79 --- /dev/null +++ b/backend/src/fintracker/sources/zenmoney/sync.py @@ -0,0 +1,169 @@ +"""The `zenmoney` source: one /v8/diff/ call per run, raw tier, then the core mapping. + +Algorithm (plan §1.7): + +1. POST /v8/diff/ with the stored cursor (`serverTimestamp`; `forceFetch` when it is 0). +2. Upsert every returned entity into `raw_zenmoney_entity` by (entity_type, id). +3. Apply `deletion`: drop the raw row, remember the fact in `raw_zenmoney_deletion`, and + flag the core row — a deleted transaction becomes `deleted` (flow_type `deleted`), a + deleted account or tag becomes archived. +4. Re-map the raw tables into the core tables (see `mapper.py`) — always from the full raw + tier, never from the diff alone, so the result does not depend on how the history was + downloaded. +5. Store the new `serverTimestamp` as the cursor. + +Re-runs are cheap and inert: an empty diff (no entities, no deletions) skips the mapper +altogether and reports `changed=False`, so the worker also skips the metrics refresh. +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import delete, update +from sqlalchemy.dialects.postgresql import insert as pg_insert + +from fintracker.models import ( + Account, + CashTxn, + Category, + FlowType, + RawZenmoneyDeletion, + RawZenmoneyEntity, +) +from fintracker.sources.base import SyncContext, SyncResult +from fintracker.sources.zenmoney import mapper +from fintracker.sources.zenmoney.client import ENTITY_TYPES, SOURCE, ZenmoneyClient + +log = logging.getLogger(__name__) + + +class ZenmoneySource: + name = SOURCE + + async def sync(self, ctx: SyncContext) -> SyncResult: + cursor_before = _as_timestamp(ctx.cursor_before) + async with ZenmoneyClient(ctx.settings, ctx.session) as client: + data = await client.diff(cursor_before) + + raw_upserted = await _upsert_entities(ctx, data) + deleted = await _apply_deletions(ctx, data) + changed = bool(raw_upserted or deleted) + + counts: dict[str, int] = {"raw_upserted": raw_upserted, "deleted": deleted} + warnings: list[str] = [] + if changed: + mapped, warnings = await mapper.map_all(ctx.session, ctx.settings) + counts.update(mapped) + await ctx.session.commit() + + cursor_after = data.get("serverTimestamp") + log.info( + "zenmoney: %s raw rows, %s deletions, cursor %s -> %s", + raw_upserted, + deleted, + cursor_before, + cursor_after, + ) + return SyncResult( + cursor_after=str(int(cursor_after)) if cursor_after is not None else None, + counts=counts, + warnings=warnings, + changed=changed, + ) + + +def _as_timestamp(cursor: str | None) -> int: + if not cursor: + return 0 + try: + return int(cursor) + except ValueError: + log.warning("zenmoney: unusable cursor %r, doing a full pull", cursor) + return 0 + + +async def _upsert_entities(ctx: SyncContext, data: dict[str, Any]) -> int: + ingested_at = datetime.now(UTC) + rows: list[dict[str, Any]] = [] + for entity_type in ENTITY_TYPES: + for payload in data.get(entity_type) or []: + if not isinstance(payload, dict) or payload.get("id") is None: + continue + rows.append( + { + "entity_type": entity_type, + "id": str(payload["id"]), + "changed": payload.get("changed"), + "payload": payload, + "ingested_at": ingested_at, + } + ) + if not rows: + return 0 + for start in range(0, len(rows), mapper.CHUNK): + chunk = rows[start : start + mapper.CHUNK] + stmt = pg_insert(RawZenmoneyEntity).values(chunk) + stmt = stmt.on_conflict_do_update( + index_elements=["entity_type", "id"], + set_={ + "changed": stmt.excluded.changed, + "payload": stmt.excluded.payload, + "ingested_at": stmt.excluded.ingested_at, + }, + ) + await ctx.session.execute(stmt) + return len(rows) + + +async def _apply_deletions(ctx: SyncContext, data: dict[str, Any]) -> int: + deletions = [d for d in (data.get("deletion") or []) if isinstance(d, dict)] + if not deletions: + return 0 + session = ctx.session + count = 0 + for item in deletions: + entity_type = str(item.get("object") or "") + entity_id = str(item.get("id") or "") + if not entity_type or not entity_id: + continue + await session.execute( + delete(RawZenmoneyEntity).where( + RawZenmoneyEntity.entity_type == entity_type, + RawZenmoneyEntity.id == entity_id, + ) + ) + stmt = pg_insert(RawZenmoneyDeletion).values( + entity_type=entity_type, + id=entity_id, + stamp=item.get("stamp"), + deleted_at=datetime.now(UTC), + ) + stmt = stmt.on_conflict_do_update( + index_elements=["entity_type", "id"], + set_={"stamp": stmt.excluded.stamp, "deleted_at": stmt.excluded.deleted_at}, + ) + await session.execute(stmt) + + if entity_type == "transaction": + await session.execute( + update(CashTxn) + .where(CashTxn.source == SOURCE, CashTxn.source_id == entity_id) + .values(deleted=True, flow_type=FlowType.deleted) + ) + elif entity_type == "account": + await session.execute( + update(Account) + .where(Account.source == SOURCE, Account.source_id == entity_id) + .values(archived=True, include_in_net_worth=False) + ) + elif entity_type == "tag": + await session.execute( + update(Category) + .where(Category.source == SOURCE, Category.source_id == entity_id) + .values(archived=True) + ) + count += 1 + return count diff --git a/backend/src/fintracker/worker/__init__.py b/backend/src/fintracker/worker/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/src/fintracker/worker/jobs.py b/backend/src/fintracker/worker/jobs.py new file mode 100644 index 0000000..3559e07 --- /dev/null +++ b/backend/src/fintracker/worker/jobs.py @@ -0,0 +1,39 @@ +"""Schedule: which source runs when (plan §2.4). Times are in the configured timezone.""" + +from __future__ import annotations + +from dataclasses import dataclass +from zoneinfo import ZoneInfo + +from apscheduler.triggers.base import BaseTrigger +from apscheduler.triggers.combining import OrTrigger +from apscheduler.triggers.cron import CronTrigger + + +@dataclass(frozen=True) +class JobSpec: + source: str + trigger: BaseTrigger + """A source appears at most once: `worker/scheduler.py` keys jobs by `sync:`, + so several times of day are one `OrTrigger`, not several specs.""" + + +MSK = ZoneInfo("Europe/Moscow") +"""CBR publishes on Moscow time, so those jobs pin the zone instead of following settings.""" + + +def default_schedule() -> list[JobSpec]: + # Each entry must name a registered source (see fintracker.sources). + return [ + JobSpec("zenmoney", CronTrigger(minute="0,30")), + # official rates appear around 13:30 MSK, the next business day's rate in the evening + JobSpec( + "cbr", + OrTrigger( + [ + CronTrigger(hour=13, minute=45, timezone=MSK), + CronTrigger(hour=18, minute=0, timezone=MSK), + ] + ), + ), + ] diff --git a/backend/src/fintracker/worker/locks.py b/backend/src/fintracker/worker/locks.py new file mode 100644 index 0000000..62dc054 --- /dev/null +++ b/backend/src/fintracker/worker/locks.py @@ -0,0 +1,36 @@ +"""Postgres advisory locks so a scheduled run and a manual trigger never overlap.""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncConnection + + +@asynccontextmanager +async def try_advisory_lock(conn: AsyncConnection, key: str) -> AsyncIterator[bool]: + """Yield True if the session-level lock for `key` was acquired; release on exit.""" + got = ( + await conn.execute(text("SELECT pg_try_advisory_lock(hashtext(:k))"), {"k": key}) + ).scalar_one() + try: + yield bool(got) + finally: + if got: + await conn.execute(text("SELECT pg_advisory_unlock(hashtext(:k))"), {"k": key}) + + +@asynccontextmanager +async def advisory_lock(conn: AsyncConnection, key: str) -> AsyncIterator[None]: + """Wait for the session-level lock for `key`, then release it on exit. + + Blocking, unlike `try_advisory_lock`: used where the work must still happen after the + current holder is done (a metric refresh queued behind a sync), not be skipped. + """ + await conn.execute(text("SELECT pg_advisory_lock(hashtext(:k))"), {"k": key}) + try: + yield + finally: + await conn.execute(text("SELECT pg_advisory_unlock(hashtext(:k))"), {"k": key}) diff --git a/backend/src/fintracker/worker/runner.py b/backend/src/fintracker/worker/runner.py new file mode 100644 index 0000000..7e70dd9 --- /dev/null +++ b/backend/src/fintracker/worker/runner.py @@ -0,0 +1,95 @@ +"""Run one source sync end to end: lock, run log, cursor, error capture.""" + +from __future__ import annotations + +import logging +import traceback +from datetime import UTC, datetime + +from sqlalchemy import select + +from fintracker.config import get_settings +from fintracker.db import get_engine, get_sessionmaker +from fintracker.metrics.refresh import refresh_all +from fintracker.models import RunStatus, SyncRun, SyncState +from fintracker.sources import registry +from fintracker.sources.base import SyncContext, SyncResult +from fintracker.worker.locks import try_advisory_lock + +log = logging.getLogger(__name__) + + +class Skipped(Exception): + """Another run of the same source holds the lock.""" + + +async def run_source(name: str, triggered_by: str) -> SyncRun: + source = registry.get(name) + settings = get_settings() + # the lock lives on its own connection for the whole run + async with ( + get_engine().connect() as lock_conn, + try_advisory_lock(lock_conn, f"sync:{name}") as got, + ): + if not got: + raise Skipped(name) + async with get_sessionmaker()() as session: + state = await session.get(SyncState, name) + if state is None: + state = SyncState(source=name) + session.add(state) + run = SyncRun( + source=name, + status=RunStatus.running, + triggered_by=triggered_by, + cursor_before=state.cursor, + ) + session.add(run) + state.last_run_at = datetime.now(UTC) + await session.commit() + run_id = run.id + + ctx = SyncContext( + session=session, + settings=settings, + cursor_before=state.cursor, + triggered_by=triggered_by, + ) + try: + result: SyncResult = await source.sync(ctx) + except Exception as exc: + await session.rollback() + log.exception("sync %s failed", name) + async with get_sessionmaker()() as s2: + run2 = await s2.get(SyncRun, run_id) + assert run2 is not None + run2.status = RunStatus.error + run2.finished_at = datetime.now(UTC) + run2.error = f"{type(exc).__name__}: {exc}\n{traceback.format_exc()[-4000:]}" + await s2.commit() + return run2 + + run = (await session.execute(select(SyncRun).where(SyncRun.id == run_id))).scalar_one() + state = await session.get(SyncState, name) + assert state is not None + if result.cursor_after is not None: + state.cursor = result.cursor_after + state.last_success_at = datetime.now(UTC) + run.status = RunStatus.ok + run.finished_at = datetime.now(UTC) + run.cursor_after = state.cursor + run.counts = dict(result.counts) + run.warnings = list(result.warnings) + await session.commit() + log.info("sync %s ok: %s", name, result.counts) + if result.changed: + entry = await refresh_all(session, trigger=f"sync:{name}") + if entry.error: + # the download itself succeeded, so the cursor stays advanced (a replay + # would fix nothing); the run is still an error, because the metrics the + # UI reads are now stale. `scheduler._finish_job` picks this up. + run.status = RunStatus.error + run.error = f"metrics refresh failed: {entry.error}" + await session.commit() + log.error("sync %s: metrics refresh failed", name) + return run diff --git a/backend/src/fintracker/worker/scheduler.py b/backend/src/fintracker/worker/scheduler.py new file mode 100644 index 0000000..9e5c03f --- /dev/null +++ b/backend/src/fintracker/worker/scheduler.py @@ -0,0 +1,118 @@ +"""Worker process: APScheduler for the timetable + a poller for manual `sync_job` rows.""" + +from __future__ import annotations + +import asyncio +import logging +import signal +from datetime import UTC, datetime + +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from sqlalchemy import select + +from fintracker.config import get_settings +from fintracker.db import get_sessionmaker, reset_engine +from fintracker.models import JobStatus, SyncJob +from fintracker.sources import registry +from fintracker.worker.jobs import default_schedule +from fintracker.worker.runner import Skipped, run_source + +log = logging.getLogger(__name__) + +POLL_SECONDS = 5 + + +async def _scheduled(source: str) -> None: + try: + await run_source(source, triggered_by="schedule") + except Skipped: + log.info("sync %s skipped: already running", source) + + +async def _process_queued_jobs() -> None: + async with get_sessionmaker()() as session: + jobs = ( + ( + await session.execute( + select(SyncJob) + .where(SyncJob.status == JobStatus.queued) + .order_by(SyncJob.requested_at) + ) + ) + .scalars() + .all() + ) + for job in jobs: + if job.source not in registry.names(): + await _finish_job(job.id, JobStatus.error, error=f"unknown source {job.source}") + continue + await _mark_running(job.id) + try: + run = await run_source(job.source, triggered_by="manual") + except Skipped: + await _finish_job(job.id, JobStatus.error, error="already running") + continue + status = JobStatus.done if run.error is None else JobStatus.error + await _finish_job(job.id, status, run_id=run.id, error=run.error) + + +async def _mark_running(job_id) -> None: + async with get_sessionmaker()() as session: + job = await session.get(SyncJob, job_id) + if job is not None: + job.status = JobStatus.running + job.started_at = datetime.now(UTC) + await session.commit() + + +async def _finish_job(job_id, status: JobStatus, *, run_id=None, error: str | None = None) -> None: + async with get_sessionmaker()() as session: + job = await session.get(SyncJob, job_id) + if job is not None: + job.status = status + job.finished_at = datetime.now(UTC) + job.run_id = run_id + job.error = error + await session.commit() + + +def build_scheduler() -> AsyncIOScheduler: + settings = get_settings() + scheduler = AsyncIOScheduler(timezone=settings.timezone) + for spec in default_schedule(): + scheduler.add_job( + _scheduled, + spec.trigger, + args=[spec.source], + id=f"sync:{spec.source}", + max_instances=1, + coalesce=True, + misfire_grace_time=600, + ) + scheduler.add_job( + _process_queued_jobs, + "interval", + seconds=POLL_SECONDS, + id="poll-sync-jobs", + max_instances=1, + coalesce=True, + ) + return scheduler + + +async def serve_forever() -> None: + scheduler = build_scheduler() + stop = asyncio.Event() + loop = asyncio.get_running_loop() + for sig in (signal.SIGINT, signal.SIGTERM): + loop.add_signal_handler(sig, stop.set) + scheduler.start() + log.info( + "worker started; sources=%s jobs=%s", registry.names(), [j.id for j in scheduler.get_jobs()] + ) + try: + await stop.wait() + finally: + log.info("worker stopping") + scheduler.shutdown(wait=True) + await reset_engine() diff --git a/backend/tests/fixtures/cbr/daily.xml b/backend/tests/fixtures/cbr/daily.xml new file mode 100644 index 0000000..14620f2 --- /dev/null +++ b/backend/tests/fixtures/cbr/daily.xml @@ -0,0 +1,27 @@ + + + +840 +USD +1 + +92,1234 +92,1234 + + +392 +JPY +100 + +61,5432 +0,615432 + + +978 +EUR +1 + +101,4567 +101,4567 + + diff --git a/backend/tests/fixtures/cbr/dynamic_jpy.xml b/backend/tests/fixtures/cbr/dynamic_jpy.xml new file mode 100644 index 0000000..09602d7 --- /dev/null +++ b/backend/tests/fixtures/cbr/dynamic_jpy.xml @@ -0,0 +1,13 @@ + + + +100 +61,5432 +0,615432 + + +100 +62,0000 +0,62 + + diff --git a/backend/tests/fixtures/cbr/dynamic_usd.xml b/backend/tests/fixtures/cbr/dynamic_usd.xml new file mode 100644 index 0000000..2173216 --- /dev/null +++ b/backend/tests/fixtures/cbr/dynamic_usd.xml @@ -0,0 +1,18 @@ + + + +1 +92,1234 +92,1234 + + +1 +92,5000 +92,5 + + +1 +93,0100 +93,01 + + diff --git a/backend/tests/fixtures/reports/raw/.gitkeep b/backend/tests/fixtures/reports/raw/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/backend/tests/fixtures/zenmoney/diff_deletion.json b/backend/tests/fixtures/zenmoney/diff_deletion.json new file mode 100644 index 0000000..7eab23a --- /dev/null +++ b/backend/tests/fixtures/zenmoney/diff_deletion.json @@ -0,0 +1,30 @@ +{ + "serverTimestamp": 1750001000, + "account": [ + { + "id": "aaaaaaaa-0000-0000-0000-000000000001", + "changed": 1750000900, + "user": 555, + "instrument": 2, + "type": "ccard", + "title": "Карта Т-Банк", + "syncID": ["4377"], + "balance": 24000.5, + "startBalance": 0, + "creditLimit": 100000, + "inBalance": true, + "savings": false, + "archive": false, + "enableCorrection": false, + "enableSMS": true + } + ], + "deletion": [ + { + "id": "dddddddd-0000-0000-0000-000000000006", + "object": "transaction", + "stamp": 1750000950, + "user": 555 + } + ] +} diff --git a/backend/tests/fixtures/zenmoney/diff_empty.json b/backend/tests/fixtures/zenmoney/diff_empty.json new file mode 100644 index 0000000..a1cdf07 --- /dev/null +++ b/backend/tests/fixtures/zenmoney/diff_empty.json @@ -0,0 +1,4 @@ +{ + "serverTimestamp": 1750000500, + "deletion": [] +} diff --git a/backend/tests/fixtures/zenmoney/diff_full.json b/backend/tests/fixtures/zenmoney/diff_full.json new file mode 100644 index 0000000..b7e8508 --- /dev/null +++ b/backend/tests/fixtures/zenmoney/diff_full.json @@ -0,0 +1,266 @@ +{ + "serverTimestamp": 1750000000, + "instrument": [ + {"id": 2, "title": "Российский рубль", "shortTitle": "RUB", "symbol": "₽", "rate": 1, "changed": 1700000000}, + {"id": 1, "title": "Доллар США", "shortTitle": "USD", "symbol": "$", "rate": 92.5, "changed": 1700000000} + ], + "company": [ + {"id": 4711, "title": "Тинькофф", "changed": 1700000000} + ], + "user": [ + {"id": 555, "country": 1, "login": "ada", "currency": 2, "changed": 1700000000} + ], + "account": [ + { + "id": "aaaaaaaa-0000-0000-0000-000000000001", + "changed": 1749000000, + "user": 555, + "instrument": 2, + "type": "ccard", + "title": "Карта Тинькофф", + "syncID": ["4377"], + "balance": 25000.5, + "startBalance": 0, + "creditLimit": 100000, + "inBalance": true, + "savings": false, + "archive": false, + "enableCorrection": false, + "enableSMS": true + }, + { + "id": "aaaaaaaa-0000-0000-0000-000000000002", + "changed": 1749000001, + "user": 555, + "instrument": 1, + "type": "cash", + "title": "Наличные USD", + "balance": 300.25, + "startBalance": 0, + "creditLimit": 0, + "inBalance": true, + "savings": false, + "archive": false, + "enableCorrection": false, + "enableSMS": false + }, + { + "id": "aaaaaaaa-0000-0000-0000-000000000003", + "changed": 1749000002, + "user": 555, + "instrument": 2, + "type": "deposit", + "title": "Вклад", + "balance": 500000, + "startBalance": 400000, + "creditLimit": 0, + "inBalance": true, + "savings": true, + "archive": false, + "enableCorrection": false, + "enableSMS": false, + "capitalization": true, + "percent": 16.5, + "startDate": "2025-02-01", + "endDateOffset": 12, + "endDateOffsetInterval": "month", + "payoffStep": 1, + "payoffInterval": "month" + }, + { + "id": "aaaaaaaa-0000-0000-0000-000000000009", + "changed": 1749000003, + "user": 555, + "instrument": 2, + "type": "debt", + "title": "Долги", + "balance": 0, + "startBalance": 0, + "creditLimit": 0, + "inBalance": false, + "savings": false, + "archive": false, + "enableCorrection": false, + "enableSMS": false + } + ], + "tag": [ + { + "id": "bbbbbbbb-0000-0000-0000-000000000001", + "changed": 1748000000, + "user": 555, + "title": "Еда", + "parent": null, + "icon": "food", + "color": 4294198070, + "showIncome": false, + "showOutcome": true, + "budgetIncome": false, + "budgetOutcome": true, + "required": false + }, + { + "id": "bbbbbbbb-0000-0000-0000-000000000002", + "changed": 1748000001, + "user": 555, + "title": "Кафе", + "parent": "bbbbbbbb-0000-0000-0000-000000000001", + "icon": "cafe", + "color": null, + "showIncome": false, + "showOutcome": true, + "budgetIncome": false, + "budgetOutcome": true, + "required": false + }, + { + "id": "bbbbbbbb-0000-0000-0000-000000000003", + "changed": 1748000002, + "user": 555, + "title": "Зарплата", + "parent": null, + "icon": null, + "color": null, + "showIncome": true, + "showOutcome": false, + "budgetIncome": true, + "budgetOutcome": false, + "required": false + } + ], + "merchant": [ + {"id": "cccccccc-0000-0000-0000-000000000001", "changed": 1748500000, "user": 555, "title": "Кофейня"} + ], + "budget": [], + "reminder": [], + "reminderMarker": [], + "transaction": [ + { + "id": "dddddddd-0000-0000-0000-000000000001", + "changed": 1749500000, + "created": 1749500000, + "user": 555, + "deleted": false, + "hold": false, + "income": 0, + "outcome": 350.4, + "incomeAccount": "aaaaaaaa-0000-0000-0000-000000000001", + "outcomeAccount": "aaaaaaaa-0000-0000-0000-000000000001", + "incomeInstrument": 2, + "outcomeInstrument": 2, + "tag": ["bbbbbbbb-0000-0000-0000-000000000002", "bbbbbbbb-0000-0000-0000-000000000001"], + "merchant": "cccccccc-0000-0000-0000-000000000001", + "payee": "COFFEE HOUSE", + "originalPayee": "COFFEE HOUSE MOSCOW", + "comment": "флэт уайт", + "date": "2025-09-01", + "mcc": 5812, + "latitude": 55.75, + "longitude": 37.61 + }, + { + "id": "dddddddd-0000-0000-0000-000000000002", + "changed": 1749500100, + "created": 1749500100, + "user": 555, + "deleted": false, + "hold": false, + "income": 180000, + "outcome": 0, + "incomeAccount": "aaaaaaaa-0000-0000-0000-000000000001", + "outcomeAccount": "aaaaaaaa-0000-0000-0000-000000000001", + "incomeInstrument": 2, + "outcomeInstrument": 2, + "tag": ["bbbbbbbb-0000-0000-0000-000000000003"], + "merchant": null, + "payee": null, + "originalPayee": null, + "comment": "зарплата за август", + "date": "2025-09-05", + "mcc": null + }, + { + "id": "dddddddd-0000-0000-0000-000000000003", + "changed": 1749500200, + "created": 1749500200, + "user": 555, + "deleted": false, + "hold": false, + "income": 50000, + "outcome": 50000, + "incomeAccount": "aaaaaaaa-0000-0000-0000-000000000003", + "outcomeAccount": "aaaaaaaa-0000-0000-0000-000000000001", + "incomeInstrument": 2, + "outcomeInstrument": 2, + "tag": null, + "merchant": null, + "payee": null, + "comment": "на вклад", + "date": "2025-09-06", + "mcc": null + }, + { + "id": "dddddddd-0000-0000-0000-000000000004", + "changed": 1749500300, + "created": 1749500300, + "user": 555, + "deleted": false, + "hold": false, + "income": 0, + "outcome": 19.99, + "incomeAccount": "aaaaaaaa-0000-0000-0000-000000000002", + "outcomeAccount": "aaaaaaaa-0000-0000-0000-000000000002", + "incomeInstrument": 1, + "outcomeInstrument": 1, + "opOutcome": 1850.5, + "opOutcomeInstrument": 2, + "tag": [], + "merchant": null, + "payee": "APPLE", + "comment": null, + "date": "2025-09-07", + "mcc": 5734 + }, + { + "id": "dddddddd-0000-0000-0000-000000000005", + "changed": 1749500400, + "created": 1749500400, + "user": 555, + "deleted": true, + "hold": false, + "income": 0, + "outcome": 999, + "incomeAccount": "aaaaaaaa-0000-0000-0000-000000000001", + "outcomeAccount": "aaaaaaaa-0000-0000-0000-000000000001", + "incomeInstrument": 2, + "outcomeInstrument": 2, + "tag": null, + "merchant": null, + "payee": "ОШИБКА", + "comment": null, + "date": "2025-09-08", + "mcc": null + }, + { + "id": "dddddddd-0000-0000-0000-000000000006", + "changed": 1749500500, + "created": 1749500500, + "user": 555, + "deleted": false, + "hold": true, + "income": 0, + "outcome": 1200, + "incomeAccount": "aaaaaaaa-0000-0000-0000-000000000001", + "outcomeAccount": "aaaaaaaa-0000-0000-0000-000000000001", + "incomeInstrument": 2, + "outcomeInstrument": 2, + "tag": null, + "merchant": null, + "payee": "АЗС", + "comment": null, + "date": "2025-09-09", + "mcc": 5541 + } + ], + "deletion": [] +} diff --git a/backend/tests/sources/__init__.py b/backend/tests/sources/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/tests/sources/conftest.py b/backend/tests/sources/conftest.py new file mode 100644 index 0000000..afc8cdf --- /dev/null +++ b/backend/tests/sources/conftest.py @@ -0,0 +1,62 @@ +"""Shared helpers for source tests: fixture loading and a SyncContext around a real session. + +Exposed as fixtures rather than importable functions: `tests/sources` is a package, and a +module named `sources` on sys.path next to `fintracker.sources` reads like a trap. +""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from pathlib import Path +from typing import Any + +import pytest +import respx + +from fintracker.config import Settings +from fintracker.db import get_sessionmaker +from fintracker.sources.base import SyncContext, SyncResult + +FIXTURES = Path(__file__).resolve().parent.parent / "fixtures" + + +@pytest.fixture +def fixture_json() -> Callable[..., dict[str, Any]]: + def _load(*parts: str) -> dict[str, Any]: + return json.loads(FIXTURES.joinpath(*parts).read_text(encoding="utf-8")) + + return _load + + +@pytest.fixture +def fixture_bytes() -> Callable[..., bytes]: + def _load(*parts: str) -> bytes: + return FIXTURES.joinpath(*parts).read_bytes() + + return _load + + +@pytest.fixture +def mock_http(): + """respx router without `assert_all_called`: some tests deliberately prove that a + route (e.g. the token endpoint) was NOT hit.""" + with respx.mock(assert_all_called=False) as router: + yield router + + +@pytest.fixture +def run_sync(): + """Run a source the way the worker does, but without the lock/run-log bookkeeping.""" + + async def _run(source: Any, *, settings: Settings, cursor: str | None = None) -> SyncResult: + async with get_sessionmaker()() as session: + ctx = SyncContext( + session=session, + settings=settings, + cursor_before=cursor, + triggered_by="test", + ) + return await source.sync(ctx) + + return _run diff --git a/backend/tests/sources/test_cbr.py b/backend/tests/sources/test_cbr.py new file mode 100644 index 0000000..c6acb2c --- /dev/null +++ b/backend/tests/sources/test_cbr.py @@ -0,0 +1,153 @@ +"""CBR source: which currencies are asked for, windows-1251 parsing, nominal handling.""" + +from __future__ import annotations + +from datetime import date, timedelta +from decimal import Decimal +from itertools import pairwise + +import httpx +from sqlalchemy import select + +from factories import make_account, make_txn +from fintracker.analytics import today_local +from fintracker.config import Settings +from fintracker.db import get_sessionmaker +from fintracker.models import RawCbrRate +from fintracker.sources.cbr.client import DAILY_URL, DYNAMIC_URL, split_range +from fintracker.sources.cbr.sync import CbrSource + +USD_ID = "R01235" +JPY_ID = "R01820" + + +def settings() -> Settings: + return Settings() + + +async def stored_rates() -> dict[tuple[str, str], tuple[int, Decimal]]: + async with get_sessionmaker()() as session: + rows = (await session.execute(select(RawCbrRate))).scalars().all() + return {(r.ccy, r.rate_date.isoformat()): (r.nominal, r.value) for r in rows} + + +def route_params(route) -> list[dict[str, str]]: + return [dict(call.request.url.params) for call in route.calls] + + +def mock_cbr(mock_http, fixture_bytes, *, dynamic: dict[str, str] | None = None): + daily = mock_http.get(DAILY_URL).mock( + return_value=httpx.Response(200, content=fixture_bytes("cbr", "daily.xml")) + ) + files = dynamic or {USD_ID: "dynamic_usd.xml", JPY_ID: "dynamic_jpy.xml"} + + def _dynamic(request: httpx.Request) -> httpx.Response: + cbr_id = request.url.params.get("VAL_NM_RQ", "") + name = files.get(cbr_id) + if name is None: + return httpx.Response(200, content=b"") + return httpx.Response(200, content=fixture_bytes("cbr", name)) + + return daily, mock_http.get(DYNAMIC_URL).mock(side_effect=_dynamic) + + +async def test_only_currencies_in_use_are_requested(app, mock_http, fixture_bytes, run_sync): + await make_account(currency="RUB") + await make_account(currency="USD", source_id="usd-cash") + await make_txn(date(2026, 9, 1), outcome="100", outcome_currency="USD") + + _, dynamic = mock_cbr(mock_http, fixture_bytes) + result = await run_sync(CbrSource(), settings=settings()) + + requested = {params["VAL_NM_RQ"] for params in route_params(dynamic)} + assert requested == {USD_ID} # EUR and JPY are quoted by CBR but unused here + assert result.counts == {"currencies": 1, "rates": 3} + assert result.warnings == [] + assert result.changed is True + assert result.cursor_after is not None + + +async def test_values_and_nominal_are_stored_as_printed(app, mock_http, fixture_bytes, run_sync): + await make_account(currency="JPY", source_id="jpy") + await make_txn(date(2026, 9, 1), outcome="1000", outcome_currency="JPY") + + mock_cbr(mock_http, fixture_bytes) + await run_sync(CbrSource(), settings=settings()) + + rates = await stored_rates() + # JPY is quoted per 100 units: both parts are kept, pricing/fx.py does the division + assert rates[("JPY", "2026-09-01")] == (100, Decimal("61.5432")) + assert rates[("JPY", "2026-09-02")] == (100, Decimal("62.0000")) + + +async def test_currency_cbr_does_not_quote_is_a_warning(app, mock_http, fixture_bytes, run_sync): + await make_account(currency="USD", source_id="usd") + await make_account(currency="XAU", source_id="gold") + await make_txn(date(2026, 9, 1), outcome="1", outcome_currency="BTC") + + _, dynamic = mock_cbr(mock_http, fixture_bytes) + result = await run_sync(CbrSource(), settings=settings()) + + assert {params["VAL_NM_RQ"] for params in route_params(dynamic)} == {USD_ID} + assert sorted(result.warnings) == [ + "BTC is not quoted by CBR — skipped (crypto or metal?)", + "XAU is not quoted by CBR — skipped (crypto or metal?)", + ] + assert result.counts["currencies"] == 1 + assert result.counts["rates"] == 3 + + +async def test_range_starts_before_the_first_transaction(app, mock_http, fixture_bytes, run_sync): + first = today_local() - timedelta(days=100) + await make_account(currency="USD", source_id="usd") + await make_txn(first, outcome="10", outcome_currency="USD") + + _, dynamic = mock_cbr(mock_http, fixture_bytes) + await run_sync(CbrSource(), settings=settings()) + + params = route_params(dynamic)[0] + assert params["date_req1"] == (first - timedelta(days=7)).strftime("%d/%m/%Y") + assert params["date_req2"] == today_local().strftime("%d/%m/%Y") + + +async def test_cursor_shortens_the_range_and_reruns_are_idempotent( + app, mock_http, fixture_bytes, run_sync +): + await make_account(currency="USD", source_id="usd") + await make_txn(today_local() - timedelta(days=10), outcome="10", outcome_currency="USD") + + _, dynamic = mock_cbr(mock_http, fixture_bytes) + first = await run_sync(CbrSource(), settings=settings()) + before = await stored_rates() + + second = await run_sync(CbrSource(), settings=settings(), cursor=first.cursor_after) + + assert second.counts == first.counts + assert await stored_rates() == before + params = route_params(dynamic)[-1] + expected_start = date.fromisoformat(first.cursor_after or "") - timedelta(days=3) + assert params["date_req1"] == expected_start.strftime("%d/%m/%Y") + + +async def test_no_foreign_currency_skips_the_network(app, mock_http, fixture_bytes, run_sync): + await make_account(currency="RUB") + daily, dynamic = mock_cbr(mock_http, fixture_bytes) + + result = await run_sync(CbrSource(), settings=settings()) + + assert not daily.called and not dynamic.called + assert result.changed is False + assert result.counts == {"currencies": 0, "rates": 0} + + +def test_long_ranges_are_chunked_by_year(): + windows = split_range(date(2020, 1, 1), date(2023, 6, 1)) + assert len(windows) == 4 + assert windows[0][0] == date(2020, 1, 1) + assert windows[-1][1] == date(2023, 6, 1) + for start, end in windows: + assert (end - start).days < 366 + # windows are contiguous, no day is fetched twice or skipped + for (_, end), (start, _) in pairwise(windows): + assert start == end + timedelta(days=1) + assert split_range(date(2026, 1, 2), date(2026, 1, 1)) == [] diff --git a/backend/tests/sources/test_zenmoney.py b/backend/tests/sources/test_zenmoney.py new file mode 100644 index 0000000..8d37d22 --- /dev/null +++ b/backend/tests/sources/test_zenmoney.py @@ -0,0 +1,517 @@ +"""ZenMoney source: the diff loop, the raw tier, the core mapping and both auth modes.""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime, timedelta +from decimal import Decimal + +import httpx +import pytest +from sqlalchemy import func, select + +from fintracker.config import Settings +from fintracker.db import get_sessionmaker +from fintracker.models import ( + Account, + AccountKind, + AccountRole, + CashTxn, + CashTxnTag, + Category, + FlowType, + Merchant, + RawZenmoneyDeletion, + RawZenmoneyEntity, + SourceCredential, + SyncState, +) +from fintracker.sources.zenmoney.client import ( + DIFF_URL, + TOKEN_URL, + ZenmoneyAuthError, +) +from fintracker.sources.zenmoney.sync import ZenmoneySource + +CARD = "aaaaaaaa-0000-0000-0000-000000000001" +USD_CASH = "aaaaaaaa-0000-0000-0000-000000000002" +DEPOSIT = "aaaaaaaa-0000-0000-0000-000000000003" +DEBT = "aaaaaaaa-0000-0000-0000-000000000009" +TAG_FOOD = "bbbbbbbb-0000-0000-0000-000000000001" +TAG_CAFE = "bbbbbbbb-0000-0000-0000-000000000002" +TXN_EXPENSE = "dddddddd-0000-0000-0000-000000000001" +TXN_INCOME = "dddddddd-0000-0000-0000-000000000002" +TXN_TRANSFER = "dddddddd-0000-0000-0000-000000000003" +TXN_USD = "dddddddd-0000-0000-0000-000000000004" +TXN_DELETED = "dddddddd-0000-0000-0000-000000000005" +TXN_HOLD = "dddddddd-0000-0000-0000-000000000006" + + +def static_settings() -> Settings: + return Settings( + zenmoney_token="static-access-token", + zenmoney_client_id=None, + zenmoney_client_secret=None, + zenmoney_refresh_token=None, + ) + + +def oauth_settings() -> Settings: + return Settings( + zenmoney_token=None, + zenmoney_client_id="client-id", + zenmoney_client_secret="client-secret", + zenmoney_refresh_token="seed-refresh-token", + ) + + +async def counts() -> dict[str, int]: + async with get_sessionmaker()() as session: + out = {} + for name, model in ( + ("raw", RawZenmoneyEntity), + ("accounts", Account), + ("categories", Category), + ("merchants", Merchant), + ("transactions", CashTxn), + ("tags", CashTxnTag), + ): + out[name] = ( + await session.execute(select(func.count()).select_from(model)) + ).scalar_one() + return out + + +async def account_by_source_id(source_id: str) -> Account: + async with get_sessionmaker()() as session: + return ( + await session.execute(select(Account).where(Account.source_id == source_id)) + ).scalar_one() + + +async def txn_by_source_id(source_id: str) -> CashTxn: + async with get_sessionmaker()() as session: + return ( + await session.execute(select(CashTxn).where(CashTxn.source_id == source_id)) + ).scalar_one() + + +async def test_first_sync_asks_for_everything_and_maps_core_rows( + app, mock_http, fixture_json, run_sync +): + diff = fixture_json("zenmoney", "diff_full.json") + route = mock_http.post(DIFF_URL).mock(return_value=httpx.Response(200, json=diff)) + + result = await run_sync(ZenmoneySource(), settings=static_settings()) + + body = json.loads(route.calls.last.request.content) + assert body["serverTimestamp"] == 0 + assert "transaction" in body["forceFetch"] and "account" in body["forceFetch"] + assert body["currentClientTimestamp"] > 1_700_000_000 + assert route.calls.last.request.headers["authorization"] == "Bearer static-access-token" + + assert result.cursor_after == "1750000000" + assert result.changed is True + assert ( + result.counts["raw_upserted"] == 18 + ) # 2 instruments + company + user + 4 acc + 3 tags + 1 merchant + 6 txns + assert result.counts["transactions"] == 6 + assert result.counts["accounts"] == 4 + assert result.counts["categories"] == 3 + + rows = await counts() + assert rows["accounts"] == 4 + assert rows["categories"] == 3 + assert rows["merchants"] == 1 + assert rows["transactions"] == 6 + assert rows["tags"] == 3 # 2 on the expense, 1 on the income + + +async def test_account_mapping(app, mock_http, fixture_json, run_sync): + mock_http.post(DIFF_URL).mock( + return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json")) + ) + await run_sync(ZenmoneySource(), settings=static_settings()) + + card = await account_by_source_id(CARD) + assert card.kind is AccountKind.zm_card # ZenMoney says "ccard" + assert card.role is AccountRole.liquid + assert card.currency == "RUB" + assert card.include_in_net_worth is True + assert card.balance == Decimal("25000.5") + assert card.credit_limit == Decimal("100000") + assert card.balance_as_of is not None + assert card.deposit_terms is None + + usd = await account_by_source_id(USD_CASH) + assert usd.currency == "USD" + assert usd.kind is AccountKind.zm_cash + + deposit = await account_by_source_id(DEPOSIT) + assert deposit.role is AccountRole.savings + assert deposit.deposit_terms is not None + assert deposit.deposit_terms["percent"] == 16.5 + assert deposit.opened_at is not None and deposit.opened_at.isoformat() == "2025-02-01" + + debt = await account_by_source_id(DEBT) + assert debt.kind is AccountKind.zm_debt + assert debt.role is AccountRole.debt + assert debt.include_in_net_worth is False # inBalance = false on the system debt account + + +async def test_category_tree_and_transaction_mapping(app, mock_http, fixture_json, run_sync): + mock_http.post(DIFF_URL).mock( + return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json")) + ) + await run_sync(ZenmoneySource(), settings=static_settings()) + + async with get_sessionmaker()() as session: + food = ( + await session.execute(select(Category).where(Category.source_id == TAG_FOOD)) + ).scalar_one() + cafe = ( + await session.execute(select(Category).where(Category.source_id == TAG_CAFE)) + ).scalar_one() + assert food.parent_id is None + assert cafe.parent_id == food.id + assert food.color == 4294198070 + + expense = ( + await session.execute(select(CashTxn).where(CashTxn.source_id == TXN_EXPENSE)) + ).scalar_one() + tags = ( + ( + await session.execute( + select(CashTxnTag) + .where(CashTxnTag.txn_id == expense.id) + .order_by(CashTxnTag.ord) + ) + ) + .scalars() + .all() + ) + + assert expense.flow_type is FlowType.expense + assert expense.outcome == Decimal("350.4") + assert expense.income == Decimal(0) + assert expense.outcome_currency == "RUB" + assert expense.mcc == 5812 + assert expense.payee == "COFFEE HOUSE" + assert expense.original_payee == "COFFEE HOUSE MOSCOW" + assert expense.merchant_id is not None + assert expense.ts == datetime.fromtimestamp(1749500000, UTC) + assert expense.date.isoformat() == "2025-09-01" + assert expense.meta == {"latitude": 55.75, "longitude": 37.61} + # primary category is the FIRST tag as ZenMoney ordered it, not the parent + assert [t.category_id for t in tags] == [cafe.id, food.id] + assert expense.primary_category_id == cafe.id + assert expense.category_id == cafe.id + + income = await txn_by_source_id(TXN_INCOME) + assert income.flow_type is FlowType.income + assert income.income == Decimal("180000") + + transfer = await txn_by_source_id(TXN_TRANSFER) + assert transfer.flow_type is FlowType.internal_transfer + assert transfer.income_account_id != transfer.outcome_account_id + assert transfer.income == Decimal("50000") and transfer.outcome == Decimal("50000") + + usd = await txn_by_source_id(TXN_USD) + assert usd.outcome_currency == "USD" + assert usd.outcome == Decimal("19.99") + assert usd.op_outcome == Decimal("1850.5") + assert usd.op_outcome_currency == "RUB" + + deleted = await txn_by_source_id(TXN_DELETED) + assert deleted.deleted is True + assert deleted.flow_type is FlowType.deleted + + hold = await txn_by_source_id(TXN_HOLD) + assert hold.hold is True + assert hold.flow_type is FlowType.expense + + +async def test_second_sync_with_empty_diff_changes_nothing(app, mock_http, fixture_json, run_sync): + mock_http.post(DIFF_URL).mock( + return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json")) + ) + first = await run_sync(ZenmoneySource(), settings=static_settings()) + before = await counts() + + mock_http.post(DIFF_URL).mock( + return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_empty.json")) + ) + second = await run_sync(ZenmoneySource(), settings=static_settings(), cursor=first.cursor_after) + + assert second.changed is False + assert second.cursor_after == "1750000500" + assert second.counts == {"raw_upserted": 0, "deleted": 0} + assert await counts() == before + + +async def test_rerunning_the_same_diff_is_idempotent(app, mock_http, fixture_json, run_sync): + mock_http.post(DIFF_URL).mock( + return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json")) + ) + await run_sync(ZenmoneySource(), settings=static_settings()) + before = await counts() + await run_sync(ZenmoneySource(), settings=static_settings()) + assert await counts() == before + + +async def test_deletion_flags_the_core_row_and_keeps_the_fact( + app, mock_http, fixture_json, run_sync +): + mock_http.post(DIFF_URL).mock( + return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json")) + ) + first = await run_sync(ZenmoneySource(), settings=static_settings()) + + mock_http.post(DIFF_URL).mock( + return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_deletion.json")) + ) + third = await run_sync(ZenmoneySource(), settings=static_settings(), cursor=first.cursor_after) + + assert third.changed is True + assert third.counts["deleted"] == 1 + assert third.cursor_after == "1750001000" + + gone = await txn_by_source_id(TXN_HOLD) + assert gone.deleted is True + assert gone.flow_type is FlowType.deleted + + card = await account_by_source_id(CARD) + # the upstream rename to "Карта Т-Банк" is NOT applied: `name` is user-owned after the + # first insert (see test_sync_does_not_overwrite_user_edits_of_an_account) + assert card.name == "Карта Тинькофф" + assert card.balance == Decimal("24000.5") + + async with get_sessionmaker()() as session: + raw = await session.get(RawZenmoneyEntity, {"entity_type": "transaction", "id": TXN_HOLD}) + assert raw is None + deletion = await session.get( + RawZenmoneyDeletion, {"entity_type": "transaction", "id": TXN_HOLD} + ) + assert deletion is not None and deletion.stamp == 1750000950 + + # the transaction count is unchanged: a deleted row is flagged, never dropped + assert (await counts())["transactions"] == 6 + + +async def test_cursor_is_stored_by_the_worker(app, mock_http, fixture_json, monkeypatch): + from fintracker.worker import runner + + monkeypatch.setattr(runner, "get_settings", static_settings) + + mock_http.post(DIFF_URL).mock( + return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json")) + ) + run = await runner.run_source("zenmoney", triggered_by="cli") + assert run.status.value == "ok", run.error + assert run.cursor_after == "1750000000" + + async with get_sessionmaker()() as session: + state = await session.get(SyncState, "zenmoney") + assert state is not None and state.cursor == "1750000000" + + +# --- auth ----------------------------------------------------------------------------- + + +async def test_static_token_401_tells_the_user_to_renew_it(app, mock_http, fixture_json, run_sync): + mock_http.post(DIFF_URL).mock(return_value=httpx.Response(401, json={"error": "nope"})) + + with pytest.raises(ZenmoneyAuthError) as exc: + await run_sync(ZenmoneySource(), settings=static_settings()) + assert "ZENMONEY_TOKEN" in str(exc.value) + + +async def test_missing_static_token_is_a_clear_error(app, run_sync): + settings = Settings( + zenmoney_token=None, + zenmoney_client_id=None, + zenmoney_client_secret=None, + zenmoney_refresh_token=None, + ) + with pytest.raises(ZenmoneyAuthError) as exc: + await run_sync(ZenmoneySource(), settings=settings) + assert "ZENMONEY_TOKEN is not set" in str(exc.value) + + +async def test_oauth_seeds_from_settings_and_stores_the_new_pair( + app, mock_http, fixture_json, run_sync +): + token_route = mock_http.post(TOKEN_URL).mock( + return_value=httpx.Response( + 200, + json={ + "access_token": "fresh-access", + "refresh_token": "fresh-refresh", + "expires_in": 86400, + "token_type": "bearer", + }, + ) + ) + diff_route = mock_http.post(DIFF_URL).mock( + return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_empty.json")) + ) + + await run_sync(ZenmoneySource(), settings=oauth_settings()) + + form = dict(httpx.QueryParams(token_route.calls.last.request.content.decode())) + assert form["grant_type"] == "refresh_token" + assert form["refresh_token"] == "seed-refresh-token" + assert form["client_id"] == "client-id" + assert form["client_secret"] == "client-secret" + assert diff_route.calls.last.request.headers["authorization"] == "Bearer fresh-access" + + async with get_sessionmaker()() as session: + cred = await session.get(SourceCredential, "zenmoney") + assert cred is not None + assert cred.payload["access_token"] == "fresh-access" + assert cred.payload["refresh_token"] == "fresh-refresh" + assert cred.payload["expires_at"] > datetime.now(UTC).isoformat() + + +async def test_oauth_refreshes_only_when_expired(app, mock_http, fixture_json, run_sync): + async with get_sessionmaker()() as session: + session.add( + SourceCredential( + source="zenmoney", + payload={ + "access_token": "still-good", + "refresh_token": "stored-refresh", + "expires_at": (datetime.now(UTC) + timedelta(hours=5)).isoformat(), + }, + ) + ) + await session.commit() + + token_route = mock_http.post(TOKEN_URL).mock(return_value=httpx.Response(200, json={})) + diff_route = mock_http.post(DIFF_URL).mock( + return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_empty.json")) + ) + + await run_sync(ZenmoneySource(), settings=oauth_settings()) + + assert not token_route.called + assert diff_route.calls.last.request.headers["authorization"] == "Bearer still-good" + + +async def test_oauth_refreshes_an_expired_token_and_retries_a_401( + app, mock_http, fixture_json, run_sync +): + async with get_sessionmaker()() as session: + session.add( + SourceCredential( + source="zenmoney", + payload={ + "access_token": "expired-access", + "refresh_token": "stored-refresh", + "expires_at": (datetime.now(UTC) - timedelta(minutes=1)).isoformat(), + }, + ) + ) + await session.commit() + + token_route = mock_http.post(TOKEN_URL).mock( + return_value=httpx.Response( + 200, + json={ + "access_token": "rotated-access", + "refresh_token": "rotated-refresh", + "expires_in": 86400, + }, + ) + ) + diff_route = mock_http.post(DIFF_URL).mock( + side_effect=[ + httpx.Response(401, json={"error": "expired"}), + httpx.Response(200, json=fixture_json("zenmoney", "diff_empty.json")), + ] + ) + + result = await run_sync(ZenmoneySource(), settings=oauth_settings()) + + assert result.cursor_after == "1750000500" + assert token_route.call_count == 2 # once because expired, once after the 401 + assert diff_route.call_count == 2 + assert diff_route.calls.last.request.headers["authorization"] == "Bearer rotated-access" + + async with get_sessionmaker()() as session: + cred = await session.get(SourceCredential, "zenmoney") + assert cred is not None and cred.payload["refresh_token"] == "rotated-refresh" + + +# --- user-owned account fields -------------------------------------------------------- + + +async def test_sync_does_not_overwrite_user_edits_of_an_account( + app, client, auth_headers, mock_http, fixture_json, run_sync +): + """`name`, `role` and `include_in_net_worth` belong to the user (PATCH /accounts/{id}); + a later sync may only refresh what the source owns, such as the balance.""" + mock_http.post(DIFF_URL).mock( + return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json")) + ) + first = await run_sync(ZenmoneySource(), settings=static_settings()) + card = await account_by_source_id(CARD) + assert card.name == "Карта Тинькофф" + + patched = await client.patch( + f"/api/v1/accounts/{card.id}", + headers=auth_headers, + json={"name": "Основная карта", "role": "savings", "include_in_net_worth": False}, + ) + assert patched.status_code == 200, patched.text + + # the same account comes back renamed and re-balanced upstream + mock_http.post(DIFF_URL).mock( + return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_deletion.json")) + ) + await run_sync(ZenmoneySource(), settings=static_settings(), cursor=first.cursor_after) + + card = await account_by_source_id(CARD) + assert card.name == "Основная карта" + assert card.role is AccountRole.savings + assert card.include_in_net_worth is False + assert card.balance == Decimal("24000.5") # source-owned: still updated + assert card.archived is False + + +async def test_archived_upstream_still_leaves_net_worth( + app, client, auth_headers, mock_http, fixture_json, run_sync +): + """The one exception to user ownership: an account the source archived cannot stay in + net worth, the same way a deleted one does not.""" + mock_http.post(DIFF_URL).mock( + return_value=httpx.Response(200, json=fixture_json("zenmoney", "diff_full.json")) + ) + first = await run_sync(ZenmoneySource(), settings=static_settings()) + card = await account_by_source_id(CARD) + assert card.include_in_net_worth is True + + patched = await client.patch( + f"/api/v1/accounts/{card.id}", headers=auth_headers, json={"role": "savings"} + ) + assert patched.status_code == 200, patched.text + + archived_diff = fixture_json("zenmoney", "diff_deletion.json") + archived_diff["account"][0]["archive"] = True + mock_http.post(DIFF_URL).mock(return_value=httpx.Response(200, json=archived_diff)) + await run_sync(ZenmoneySource(), settings=static_settings(), cursor=first.cursor_after) + + card = await account_by_source_id(CARD) + assert card.archived is True + assert card.include_in_net_worth is False + assert card.role is AccountRole.savings # still the user's choice + + +async def test_instrument_id_zero_is_a_valid_currency(app): + from fintracker.sources.zenmoney.mapper import _ccy, instrument_codes + + codes = instrument_codes([{"id": 0, "shortTitle": "RUB"}, {"id": 2, "shortTitle": "USD"}]) + assert _ccy(codes, 0) == "RUB" + assert _ccy(codes, 2) == "USD" + assert _ccy(codes, None) is None + assert _ccy(codes, 99) is None diff --git a/backend/tests/test_sync.py b/backend/tests/test_sync.py new file mode 100644 index 0000000..d351f86 --- /dev/null +++ b/backend/tests/test_sync.py @@ -0,0 +1,115 @@ +import pytest + +from fintracker.sources import registry +from fintracker.sources.base import SyncContext, SyncResult + + +class FakeSource: + name = "fake" + + def __init__(self) -> None: + self.calls = 0 + + async def sync(self, ctx: SyncContext) -> SyncResult: + self.calls += 1 + before = int(ctx.cursor_before or 0) + return SyncResult(cursor_after=str(before + 1), counts={"rows": 3}) + + +class FailingSource: + name = "boom" + + async def sync(self, ctx: SyncContext) -> SyncResult: + raise RuntimeError("upstream down") + + +@pytest.fixture +def fake_source(): + src = registry.register(FakeSource()) + registry.register(FailingSource()) + yield src + registry.unregister("fake") + registry.unregister("boom") + + +async def test_trigger_dedupes_queued_jobs(client, auth_headers, fake_source): + r1 = await client.post("/api/v1/sync/fake", headers=auth_headers) + r2 = await client.post("/api/v1/sync/fake", headers=auth_headers) + assert r1.status_code == 202 and r2.status_code == 202 + assert r1.json()["id"] == r2.json()["id"] + + r = await client.post("/api/v1/sync/unknown", headers=auth_headers) + assert r.status_code == 404 + + status = await client.get("/api/v1/sync/status", headers=auth_headers) + assert status.status_code == 200 + by_name = {s["source"]: s for s in status.json()} + assert by_name["fake"]["queued"] is True + assert by_name["fake"]["cursor"] is None + + +async def test_worker_runs_queued_job_and_advances_cursor(client, auth_headers, fake_source): + from fintracker.worker.scheduler import _process_queued_jobs + + await client.post("/api/v1/sync/fake", headers=auth_headers) + await _process_queued_jobs() + assert fake_source.calls == 1 + + status = await client.get("/api/v1/sync/status", headers=auth_headers) + fake = next(s for s in status.json() if s["source"] == "fake") + assert fake["cursor"] == "1" + assert fake["queued"] is False + assert fake["last_run_status"] == "ok" + + runs = await client.get("/api/v1/sync/runs", params={"source": "fake"}, headers=auth_headers) + assert runs.json()[0]["counts"] == {"rows": 3} + assert runs.json()[0]["triggered_by"] == "manual" + + +async def test_failed_sync_is_recorded_not_raised(client, auth_headers, fake_source): + from fintracker.worker.runner import run_source + + run = await run_source("boom", triggered_by="cli") + assert run.status.value == "error" + assert run.error is not None and "upstream down" in run.error + + status = await client.get("/api/v1/sync/status", headers=auth_headers) + boom = next(s for s in status.json() if s["source"] == "boom") + assert boom["last_run_status"] == "error" + assert boom["last_success_at"] is None + + +async def test_metrics_failure_after_a_sync_marks_the_run_as_error(app, fake_source): + """The download succeeded, so the cursor stays advanced; the run is still an error, + because the metrics the UI reads are stale.""" + from fintracker.db import get_sessionmaker + from fintracker.metrics import refresh as refresh_module + from fintracker.models import SyncState + from fintracker.worker.runner import run_source + + async def failing_step(session) -> None: + raise RuntimeError("step exploded") + + refresh_module.STEPS.append(("test_failing_step", failing_step)) + try: + run = await run_source("fake", triggered_by="cli") + finally: + refresh_module.STEPS[:] = [s for s in refresh_module.STEPS if s[0] != "test_failing_step"] + + assert run.status.value == "error" + assert run.error is not None + assert run.error.startswith("metrics refresh failed:") + assert "step exploded" in run.error + # the sync itself succeeded: the cursor must not replay + assert run.cursor_after == "1" + async with get_sessionmaker()() as session: + state = await session.get(SyncState, "fake") + assert state is not None and state.cursor == "1" + + +async def test_a_later_refresh_is_clean_again(app, fake_source): + """The temporary failing step is gone, so nothing leaks into other tests.""" + from fintracker.worker.runner import run_source + + run = await run_source("fake", triggered_by="cli") + assert run.status.value == "ok", run.error