feat(backend): каркас — конфиг, движок БД, базовые модели и первая миграция

SQLAlchemy 2 async на asyncpg, Alembic, pydantic-settings. Деньги везде
NUMERIC(24,10) с колонкой валюты рядом (db/base.py), postgres-enum'ы хранят
значения, а не имена, чтобы база читалась так же, как API.

Первая миграция: app_user, account, portfolio, instrument, sync_run, sync_job.
metrics/refresh.py задаёт порядок пересборки metric_* и сериализует параллельные
пересчёты advisory-локом на отдельном соединении: каждый шаг заменяет свою
таблицу целиком, два одновременных прогона затёрли бы друг друга.
This commit is contained in:
Dmitry
2026-09-18 13:43:31 +03:00
parent 220f027652
commit 295438914d
19 changed files with 3406 additions and 0 deletions
+39
View File
@@ -0,0 +1,39 @@
[alembic]
script_location = alembic
prepend_sys_path = src
path_separator = os
file_template = %%(rev)s_%%(slug)s
# DATABASE_URL from the environment overrides this (see alembic/env.py)
sqlalchemy.url = postgresql+asyncpg://fintracker:fintracker@localhost:54329/fintracker
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
+59
View File
@@ -0,0 +1,59 @@
"""Alembic environment: async engine, DATABASE_URL from the environment wins over alembic.ini."""
from __future__ import annotations
import asyncio
import os
from logging.config import fileConfig
from alembic import context
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config
import fintracker.models # noqa: F401 — register every table on Base.metadata
from fintracker.db.base import Base
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
if url := os.environ.get("DATABASE_URL"):
config.set_main_option("sqlalchemy.url", url)
target_metadata = Base.metadata
def run_migrations_offline() -> None:
context.configure(
url=config.get_main_option("sqlalchemy.url"),
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
compare_type=True,
)
with context.begin_transaction():
context.run_migrations()
def _run_sync(connection: Connection) -> None:
context.configure(connection=connection, target_metadata=target_metadata, compare_type=True)
with context.begin_transaction():
context.run_migrations()
async def run_migrations_online() -> None:
connectable = async_engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
async with connectable.connect() as connection:
await connection.run_sync(_run_sync)
await connectable.dispose()
if context.is_offline_mode():
run_migrations_offline()
else:
asyncio.run(run_migrations_online())
+27
View File
@@ -0,0 +1,27 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
${imports if imports else ""}
revision: str = ${repr(up_revision)}
down_revision: str | None = ${repr(down_revision)}
branch_labels: str | Sequence[str] | None = ${repr(branch_labels)}
depends_on: str | Sequence[str] | None = ${repr(depends_on)}
def upgrade() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}
@@ -0,0 +1,401 @@
"""initial
Revision ID: 05beabc436ad
Revises:
Create Date: 2026-09-17 22:02:22.073683
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
revision: str = "05beabc436ad"
down_revision: str | None = None
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(
"account",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column(
"kind",
sa.Enum(
"zm_cash",
"zm_card",
"zm_checking",
"zm_deposit",
"zm_loan",
"zm_emoney",
"zm_debt",
"broker",
"manual_asset",
name="account_kind",
),
nullable=False,
),
sa.Column("source", sa.String(length=32), nullable=False),
sa.Column("source_id", sa.String(length=128), nullable=False),
sa.Column(
"broker", sa.Enum("tinvest", "sber", "vtb", "other", name="broker"), nullable=True
),
sa.Column("name", sa.String(length=256), nullable=False),
sa.Column("currency", sa.String(length=3), nullable=False),
sa.Column("include_in_net_worth", sa.Boolean(), nullable=False),
sa.Column(
"role",
sa.Enum("liquid", "savings", "investment", "debt", name="account_role"),
nullable=False,
),
sa.Column("mirror_of_account_id", sa.Integer(), nullable=True),
sa.Column(
"primary_event_source",
sa.Enum("tinvest_api", "report_sber", "report_vtb", "manual", name="event_source"),
nullable=True,
),
sa.Column("opened_at", sa.Date(), nullable=True),
sa.Column("archived", sa.Boolean(), nullable=False),
sa.Column("deposit_terms", 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(
["mirror_of_account_id"],
["account.id"],
name=op.f("fk_account_mirror_of_account_id_account"),
ondelete="SET NULL",
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_account")),
sa.UniqueConstraint("source", "source_id", name=op.f("uq_account_source_source_id")),
)
op.create_table(
"app_user",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("email", sa.String(length=320), nullable=False),
sa.Column("password_hash", sa.String(length=512), nullable=False),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
server_default=sa.text("now()"),
nullable=False,
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_app_user")),
sa.UniqueConstraint("email", name=op.f("uq_app_user_email")),
)
op.create_table(
"instrument",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column(
"asset_class",
sa.Enum(
"share",
"bond",
"etf",
"fund",
"currency",
"index",
"deposit",
"real_estate",
"crypto",
"custom",
name="asset_class",
),
nullable=False,
),
sa.Column("isin", sa.String(length=12), nullable=True),
sa.Column("figi", sa.String(length=12), nullable=True),
sa.Column("tinvest_uid", sa.String(length=64), nullable=True),
sa.Column("ticker", sa.String(length=32), nullable=True),
sa.Column("board", sa.String(length=16), nullable=True),
sa.Column("exchange", sa.String(length=32), nullable=True),
sa.Column("name", sa.String(length=256), nullable=False),
sa.Column("issuer", sa.String(length=256), nullable=True),
sa.Column("currency", sa.String(length=3), nullable=False),
sa.Column("lot", sa.Integer(), nullable=False),
sa.Column("nominal", sa.Numeric(precision=24, scale=10), nullable=True),
sa.Column("nominal_currency", sa.String(length=3), nullable=True),
sa.Column("maturity_date", sa.Date(), nullable=True),
sa.Column("sector", sa.String(length=64), nullable=True),
sa.Column("country", sa.String(length=2), nullable=True),
sa.Column("is_active", sa.Boolean(), nullable=False),
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.PrimaryKeyConstraint("id", name=op.f("pk_instrument")),
)
op.create_index(
"uq_instrument_figi",
"instrument",
["figi"],
unique=True,
postgresql_where=sa.text("figi IS NOT NULL"),
)
op.create_index(
"uq_instrument_isin",
"instrument",
["isin"],
unique=True,
postgresql_where=sa.text("isin IS NOT NULL"),
)
op.create_index(
"uq_instrument_ticker_board",
"instrument",
["ticker", "board"],
unique=True,
postgresql_where=sa.text("ticker IS NOT NULL AND board IS NOT NULL"),
)
op.create_index(
"uq_instrument_tinvest_uid",
"instrument",
["tinvest_uid"],
unique=True,
postgresql_where=sa.text("tinvest_uid IS NOT NULL"),
)
op.create_table(
"portfolio",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("name", sa.String(length=128), nullable=False),
sa.Column("is_default", sa.Boolean(), nullable=False),
sa.Column("base_currency", sa.String(length=3), 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_portfolio")),
sa.UniqueConstraint("name", name=op.f("uq_portfolio_name")),
)
op.create_table(
"source_credential",
sa.Column("source", sa.String(length=64), nullable=False),
sa.Column("payload", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
sa.Column(
"updated_at",
sa.DateTime(timezone=True),
server_default=sa.text("now()"),
nullable=False,
),
sa.PrimaryKeyConstraint("source", name=op.f("pk_source_credential")),
)
op.create_table(
"sync_run",
sa.Column("id", sa.Uuid(), nullable=False),
sa.Column("source", sa.String(length=64), nullable=False),
sa.Column("status", sa.Enum("running", "ok", "error", name="run_status"), nullable=False),
sa.Column("triggered_by", sa.String(length=32), 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("cursor_before", sa.Text(), nullable=True),
sa.Column("cursor_after", sa.Text(), nullable=True),
sa.Column("counts", postgresql.JSONB(astext_type=sa.Text()), nullable=True),
sa.Column("warnings", postgresql.JSONB(astext_type=sa.Text()), nullable=True),
sa.Column("error", sa.Text(), nullable=True),
sa.PrimaryKeyConstraint("id", name=op.f("pk_sync_run")),
)
op.create_index(op.f("ix_sync_run_source"), "sync_run", ["source"], unique=False)
op.create_table(
"sync_state",
sa.Column("source", sa.String(length=64), nullable=False),
sa.Column("cursor", sa.Text(), nullable=True),
sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("last_success_at", sa.DateTime(timezone=True), nullable=True),
sa.PrimaryKeyConstraint("source", name=op.f("pk_sync_state")),
)
op.create_table(
"account_link",
sa.Column("zm_account_id", sa.Integer(), nullable=False),
sa.Column("broker_account_id", sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(
["broker_account_id"],
["account.id"],
name=op.f("fk_account_link_broker_account_id_account"),
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["zm_account_id"],
["account.id"],
name=op.f("fk_account_link_zm_account_id_account"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("zm_account_id", name=op.f("pk_account_link")),
sa.UniqueConstraint("broker_account_id", name=op.f("uq_account_link_broker_account_id")),
)
op.create_table(
"instrument_alias",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("instrument_id", sa.Integer(), nullable=False),
sa.Column("source", sa.String(length=32), nullable=False),
sa.Column("source_key", sa.String(length=256), nullable=False),
sa.ForeignKeyConstraint(
["instrument_id"],
["instrument.id"],
name=op.f("fk_instrument_alias_instrument_id_instrument"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_instrument_alias")),
sa.UniqueConstraint(
"source", "source_key", name=op.f("uq_instrument_alias_source_source_key")
),
)
op.create_index(
op.f("ix_instrument_alias_instrument_id"),
"instrument_alias",
["instrument_id"],
unique=False,
)
op.create_table(
"portfolio_account",
sa.Column("portfolio_id", sa.Integer(), nullable=False),
sa.Column("account_id", sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(
["account_id"],
["account.id"],
name=op.f("fk_portfolio_account_account_id_account"),
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["portfolio_id"],
["portfolio.id"],
name=op.f("fk_portfolio_account_portfolio_id_portfolio"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("portfolio_id", "account_id", name=op.f("pk_portfolio_account")),
)
op.create_table(
"refresh_token",
sa.Column("id", sa.Uuid(), nullable=False),
sa.Column("user_id", sa.Integer(), nullable=False),
sa.Column("token_hash", sa.String(length=64), nullable=False),
sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
server_default=sa.text("now()"),
nullable=False,
),
sa.ForeignKeyConstraint(
["user_id"],
["app_user.id"],
name=op.f("fk_refresh_token_user_id_app_user"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_refresh_token")),
sa.UniqueConstraint("token_hash", name=op.f("uq_refresh_token_token_hash")),
)
op.create_index(op.f("ix_refresh_token_user_id"), "refresh_token", ["user_id"], unique=False)
op.create_table(
"sync_job",
sa.Column("id", sa.Uuid(), nullable=False),
sa.Column("source", sa.String(length=64), nullable=False),
sa.Column(
"status",
sa.Enum("queued", "running", "done", "error", name="job_status"),
nullable=False,
),
sa.Column(
"requested_at",
sa.DateTime(timezone=True),
server_default=sa.text("now()"),
nullable=False,
),
sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("run_id", sa.Uuid(), nullable=True),
sa.Column("error", sa.Text(), nullable=True),
sa.ForeignKeyConstraint(
["run_id"],
["sync_run.id"],
name=op.f("fk_sync_job_run_id_sync_run"),
ondelete="SET NULL",
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_sync_job")),
)
op.create_index(op.f("ix_sync_job_source"), "sync_job", ["source"], unique=False)
op.create_index(op.f("ix_sync_job_status"), "sync_job", ["status"], unique=False)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f("ix_sync_job_status"), table_name="sync_job")
op.drop_index(op.f("ix_sync_job_source"), table_name="sync_job")
op.drop_table("sync_job")
op.drop_index(op.f("ix_refresh_token_user_id"), table_name="refresh_token")
op.drop_table("refresh_token")
op.drop_table("portfolio_account")
op.drop_index(op.f("ix_instrument_alias_instrument_id"), table_name="instrument_alias")
op.drop_table("instrument_alias")
op.drop_table("account_link")
op.drop_table("sync_state")
op.drop_index(op.f("ix_sync_run_source"), table_name="sync_run")
op.drop_table("sync_run")
op.drop_table("source_credential")
op.drop_table("portfolio")
op.drop_index(
"uq_instrument_tinvest_uid",
table_name="instrument",
postgresql_where=sa.text("tinvest_uid IS NOT NULL"),
)
op.drop_index(
"uq_instrument_ticker_board",
table_name="instrument",
postgresql_where=sa.text("ticker IS NOT NULL AND board IS NOT NULL"),
)
op.drop_index(
"uq_instrument_isin", table_name="instrument", postgresql_where=sa.text("isin IS NOT NULL")
)
op.drop_index(
"uq_instrument_figi", table_name="instrument", postgresql_where=sa.text("figi IS NOT NULL")
)
op.drop_table("instrument")
op.drop_table("app_user")
op.drop_table("account")
# ### end Alembic commands ###
for enum_name in (
"account_kind",
"broker",
"account_role",
"event_source",
"asset_class",
"run_status",
"job_status",
):
sa.Enum(name=enum_name).drop(op.get_bind(), checkfirst=True)
+82
View File
@@ -0,0 +1,82 @@
[project]
name = "fintracker"
version = "0.1.0"
description = "Personal finance + investment analytics: ZenMoney and brokers in one place"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"fastapi>=0.115",
"uvicorn[standard]>=0.30",
"sqlalchemy[asyncio]>=2.0.36",
"asyncpg>=0.30",
"alembic>=1.14",
"pydantic[email]>=2.9",
"pydantic-settings>=2.5",
"httpx[socks]>=0.27",
"pyjwt>=2.9",
"pwdlib[argon2]>=0.2",
"typer>=0.12",
"apscheduler>=3.10,<4",
"python-multipart>=0.0.12",
"t-tech-investments>=1.51.0",
"pyxirr>=0.10.8",
]
[project.scripts]
fintracker = "fintracker.cli:app"
[dependency-groups]
dev = [
"pytest>=8.3",
"pytest-asyncio>=0.24",
"pytest-postgresql>=6.1",
"psycopg[binary]>=3.2",
"ruff>=0.7",
"pyright>=1.1.380",
"respx>=0.21",
]
# T-Invest SDK lives on T-Bank's own index (public PyPI copy is quarantined).
[[tool.uv.index]]
name = "tbank-invest"
url = "https://opensource.tbank.ru/api/v4/projects/238/packages/pypi/simple"
explicit = true
[tool.uv.sources]
t-tech-investments = { index = "tbank-invest" }
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["src/fintracker"]
[tool.ruff]
line-length = 100
src = ["src", "tests"]
target-version = "py312"
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM", "RUF"]
ignore = ["RUF001", "RUF002", "RUF003"] # cyrillic in strings/comments is intended
[tool.ruff.lint.per-file-ignores]
"tests/*" = ["E501"]
"alembic/versions/*" = ["E501"]
[tool.pyright]
venvPath = "."
venv = ".venv"
include = ["src", "tests"]
extraPaths = ["src"]
pythonVersion = "3.12"
typeCheckingMode = "standard"
reportMissingImports = "warning"
[tool.pytest.ini_options]
pythonpath = ["src"]
testpaths = ["tests"]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"
asyncio_default_test_loop_scope = "session"
+3
View File
@@ -0,0 +1,3 @@
"""fintracker — personal finance + investment analytics backend."""
__version__ = "0.1.0"
+140
View File
@@ -0,0 +1,140 @@
"""`fintracker` command line: serve / worker / create-user / openapi / sync."""
from __future__ import annotations
import asyncio
import json
import logging
from pathlib import Path
from typing import Annotated
import typer
from fintracker.config import get_settings
app = typer.Typer(no_args_is_help=True, add_completion=False)
def _logging() -> None:
logging.basicConfig(
level=get_settings().log_level.upper(),
format="%(asctime)s %(levelname)-5s %(name)s: %(message)s",
)
@app.command()
def serve(
host: str = "127.0.0.1",
port: int = 8000,
reload: bool = typer.Option(False, "--reload", help="dev only"),
) -> None:
"""Run the HTTP API."""
import uvicorn
uvicorn.run("fintracker.api.app:create_app", factory=True, host=host, port=port, reload=reload)
@app.command()
def worker() -> None:
"""Run the scheduled-sync worker (one instance per deployment)."""
_logging()
from fintracker.worker.scheduler import serve_forever
asyncio.run(serve_forever())
@app.command("create-user")
def create_user(
email: str,
password: str = typer.Option(..., prompt=True, hide_input=True, confirmation_prompt=True),
) -> None:
"""Create (or reset the password of) the application user."""
from sqlalchemy import select
from fintracker.api.security import hash_password
from fintracker.db import get_sessionmaker, reset_engine
from fintracker.models import AppUser
async def _run() -> None:
async with get_sessionmaker()() as session:
user = (
await session.execute(select(AppUser).where(AppUser.email == email.lower()))
).scalar_one_or_none()
if user is None:
session.add(AppUser(email=email.lower(), password_hash=hash_password(password)))
action = "created"
else:
user.password_hash = hash_password(password)
action = "password reset"
await session.commit()
await reset_engine()
typer.echo(f"{email.lower()}: {action}")
asyncio.run(_run())
@app.command()
def openapi(out: Annotated[Path, typer.Argument()] = Path("openapi.json")) -> None:
"""Write the OpenAPI schema to a file (committed as openapi/openapi.json)."""
from fintracker.api.app import create_app
schema = create_app().openapi()
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(schema, indent=2, ensure_ascii=False, sort_keys=True) + "\n")
typer.echo(f"wrote {out}")
@app.command()
def sync(source: str) -> None:
"""Run one source sync in-process (uses the same lock as the worker)."""
_logging()
from fintracker.db import reset_engine
from fintracker.sources import registry
from fintracker.worker.runner import Skipped, run_source
if source not in registry.names():
raise typer.BadParameter(f"unknown source {source!r}; known: {registry.names()}")
async def _run() -> None:
try:
run = await run_source(source, triggered_by="cli")
except Skipped:
typer.echo(f"{source}: already running elsewhere")
raise typer.Exit(1) from None
finally:
await reset_engine()
typer.echo(f"{source}: {run.status.value} counts={run.counts} warnings={run.warnings}")
if run.error:
typer.echo(run.error, err=True)
raise typer.Exit(1)
asyncio.run(_run())
metrics_app = typer.Typer(help="Precomputed metrics.")
app.add_typer(metrics_app, name="metrics")
@metrics_app.command("refresh")
def metrics_refresh() -> None:
"""Rebuild every metric_* table from core data (idempotent)."""
_logging()
from fintracker.db import get_sessionmaker, reset_engine
from fintracker.metrics.refresh import refresh_all
async def _run() -> None:
try:
async with get_sessionmaker()() as session:
entry = await refresh_all(session, trigger="cli")
finally:
await reset_engine()
if entry.error:
typer.echo(entry.error, err=True)
raise typer.Exit(1)
typer.echo(f"metrics refreshed at {entry.finished_at}")
asyncio.run(_run())
if __name__ == "__main__":
app()
+79
View File
@@ -0,0 +1,79 @@
"""Runtime settings, read once from the environment (and a local .env in dev)."""
from __future__ import annotations
from functools import lru_cache
from pathlib import Path
from pydantic import field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
_REPO_ROOT = Path(__file__).resolve().parents[3]
"""Repo root: backend/src/fintracker/config.py -> ../../.. — the dev `.env` lives there."""
class Settings(BaseSettings):
model_config = SettingsConfigDict(
# absolute first: recipes run from `backend/`, so a bare ".env" would never resolve.
# A ".env" next to the process (docker image, other layouts) still wins if present.
env_file=(_REPO_ROOT / ".env", ".env"),
env_file_encoding="utf-8",
extra="ignore",
)
database_url: str = "postgresql+asyncpg://fintracker:fintracker@localhost:54329/fintracker"
"""SQLAlchemy async URL. Inside docker compose the host is `db`."""
jwt_secret: str = "dev-insecure-secret-change-me"
access_token_ttl_seconds: int = 3600
refresh_token_ttl_seconds: int = 30 * 86400
login_rate_limit_per_minute: int = 5
cors_origins: str = ""
"""Comma-separated origins for the Flutter web build when it is NOT served by Caddy
from the same origin (e.g. `flutter run -d chrome` during development)."""
timezone: str = "Europe/Moscow"
uploads_dir: Path = Path("/data/uploads")
web_dir: Path | None = None
"""Optional Flutter web build to serve at `/` (dev convenience; Caddy does this in prod)."""
log_level: str = "INFO"
base_currency: str = "RUB"
"""Display/metric currency; native amounts are always kept alongside."""
# --- sources: secrets from .env; rotating refresh tokens live in source_credential ---
zenmoney_token: str | None = None
"""Static access token (e.g. from zerro.app/token). Used when no OAuth client is configured."""
zenmoney_client_id: str | None = None
zenmoney_client_secret: str | None = None
zenmoney_refresh_token: str | None = None
"""Initial refresh token for OAuth rotation; later values are stored in source_credential."""
tinvest_token: str | None = None
@field_validator(
"web_dir",
"zenmoney_token",
"zenmoney_client_id",
"zenmoney_client_secret",
"zenmoney_refresh_token",
"tinvest_token",
mode="before",
)
@classmethod
def _blank_is_none(cls, v: object) -> object:
"""A blank line in `.env` (`WEB_DIR=`) means unset — not Path('.'), not an empty token."""
return None if isinstance(v, str) and not v.strip() else v
@property
def cors_origin_list(self) -> list[str]:
return [o.strip() for o in self.cors_origins.split(",") if o.strip()]
@property
def is_dev_secret(self) -> bool:
return self.jwt_secret.startswith("dev-insecure")
@lru_cache
def get_settings() -> Settings:
return Settings()
+4
View File
@@ -0,0 +1,4 @@
from fintracker.db.base import Base, TimestampMixin
from fintracker.db.engine import get_engine, get_sessionmaker, reset_engine
__all__ = ["Base", "TimestampMixin", "get_engine", "get_sessionmaker", "reset_engine"]
+46
View File
@@ -0,0 +1,46 @@
"""Declarative base shared by every model.
Conventions (see plan §1): money and quantities are NUMERIC(24,10) with a currency
column next to them; timestamps are timezone-aware; JSON payloads are JSONB.
"""
from __future__ import annotations
import enum
from datetime import datetime
from decimal import Decimal
from typing import Any
from sqlalchemy import DateTime, Enum, MetaData, Numeric, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
NAMING_CONVENTION = {
"ix": "ix_%(column_0_label)s",
"uq": "uq_%(table_name)s_%(column_0_N_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_N_name)s_%(referred_table_name)s",
"pk": "pk_%(table_name)s",
}
MONEY = Numeric(24, 10)
class Base(DeclarativeBase):
metadata = MetaData(naming_convention=NAMING_CONVENTION)
type_annotation_map = { # noqa: RUF012 — SQLAlchemy reads this class attribute
Decimal: MONEY,
datetime: DateTime(timezone=True),
dict[str, Any]: JSONB,
list[Any]: JSONB,
}
class TimestampMixin:
created_at: Mapped[datetime] = mapped_column(server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(server_default=func.now(), onupdate=func.now())
def db_enum(enum_cls: type[enum.Enum], name: str) -> Enum:
"""Postgres enum that stores member *values* (not names), so the DB reads like the API."""
return Enum(enum_cls, name=name, values_callable=lambda e: [m.value for m in e])
+42
View File
@@ -0,0 +1,42 @@
"""Lazily created async engine + session factory.
Created on first use (not at import) so tests and the CLI can point DATABASE_URL
anywhere before anything touches the database. `reset_engine()` drops the cache.
"""
from __future__ import annotations
from sqlalchemy.ext.asyncio import (
AsyncEngine,
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from fintracker.config import get_settings
_engine: AsyncEngine | None = None
_sessionmaker: async_sessionmaker[AsyncSession] | None = None
def get_engine() -> AsyncEngine:
global _engine
if _engine is None:
_engine = create_async_engine(get_settings().database_url, pool_pre_ping=True)
return _engine
def get_sessionmaker() -> async_sessionmaker[AsyncSession]:
global _sessionmaker
if _sessionmaker is None:
_sessionmaker = async_sessionmaker(get_engine(), expire_on_commit=False)
return _sessionmaker
async def reset_engine() -> None:
"""Dispose the cached engine (tests, or after settings change)."""
global _engine, _sessionmaker
if _engine is not None:
await _engine.dispose()
_engine = None
_sessionmaker = None
+66
View File
@@ -0,0 +1,66 @@
"""Rebuild every metric_* table after data changed (plan §3).
Order (each step is a function taking an AsyncSession and committing its own tables):
fx -> classify -> net worth -> cash flow -> spending -> runway -> data quality
Phase 2 inserts prices/lots/valuation/holdings/returns between fx and net worth.
`refresh_all` is what the worker calls after a sync that reported `changed=True`, what the
CLI `fintracker metrics refresh` runs, and what `POST /metrics/refresh` queues.
Every step replaces its whole table, so two refreshes at once would race (a unique violation
at best, half of one run's rows at worst). A session-level Postgres advisory lock on its own
connection serialises them: a refresh that arrives during a sync WAITS and then runs, because
`POST /rules/apply` must take effect, not be silently dropped.
"""
from __future__ import annotations
import logging
import traceback
from collections.abc import Awaitable, Callable
from datetime import UTC, datetime
from sqlalchemy.ext.asyncio import AsyncSession
from fintracker.models import MetricRefreshLog
log = logging.getLogger(__name__)
LOCK_KEY = "metrics:refresh"
Step = Callable[[AsyncSession], Awaitable[None]]
STEPS: list[tuple[str, Step]] = []
"""Filled by analytics modules via `register_step`; order of registration == run order."""
def register_step(name: str, step: Step) -> None:
STEPS.append((name, step))
async def refresh_all(session: AsyncSession, trigger: str) -> MetricRefreshLog:
# imported here (not at module scope) so the analytics import graph stays out of the
# worker/CLI import path until a refresh actually runs; registration is idempotent
from fintracker.analytics import register_steps
from fintracker.db import get_engine
from fintracker.worker.locks import advisory_lock
register_steps()
# the lock lives on its own connection, so it is independent of `session`'s transactions
async with get_engine().connect() as lock_conn, advisory_lock(lock_conn, LOCK_KEY):
entry = MetricRefreshLog(trigger=trigger)
session.add(entry)
await session.commit()
try:
for name, step in STEPS:
log.info("metrics: %s", name)
await step(session)
await session.commit()
except Exception as exc:
await session.rollback()
entry.error = f"{type(exc).__name__}: {exc}\n{traceback.format_exc()[-4000:]}"
log.exception("metrics refresh failed at step")
entry.finished_at = datetime.now(UTC)
await session.commit()
return entry
+136
View File
@@ -0,0 +1,136 @@
"""ORM models. Importing this package registers every table on `Base.metadata`."""
from fintracker.models.accounts import (
Account,
AccountKind,
AccountLink,
AccountRole,
Broker,
EventSource,
Portfolio,
PortfolioAccount,
)
from fintracker.models.auth import AppUser, RefreshToken
from fintracker.models.instruments import AssetClass, Instrument, InstrumentAlias
from fintracker.models.ledger import (
EXTERNAL_FLOW_KINDS,
POSITION_KINDS,
Event,
EventKind,
EventStatus,
Lot,
LotDisposal,
)
from fintracker.models.metrics import (
MetricCashFlowMonthly,
MetricDataQuality,
MetricHolding,
MetricNetWorthDaily,
MetricPortfolioValueDaily,
MetricRefreshLog,
MetricReturns,
MetricRunway,
MetricSpendingByCategory,
)
from fintracker.models.pricing import (
CashSnapshot,
CorporateAction,
CorporateActionKind,
CorporateActionStatus,
FxRateDaily,
PositionSnapshot,
PriceDaily,
PriceLast,
PriceManual,
RawCbrRate,
)
from fintracker.models.sync import (
JobStatus,
RunStatus,
SourceCredential,
SyncJob,
SyncRun,
SyncState,
)
from fintracker.models.tinvest import (
RawTinvestEvent,
RawTinvestInstrument,
RawTinvestOperation,
RawTinvestSnapshot,
)
from fintracker.models.zenmoney import (
CashTxn,
CashTxnTag,
Category,
FlowType,
Merchant,
RawZenmoneyDeletion,
RawZenmoneyEntity,
Rule,
RuleKind,
RuleMatchType,
Trip,
)
__all__ = [
"EXTERNAL_FLOW_KINDS",
"POSITION_KINDS",
"Account",
"AccountKind",
"AccountLink",
"AccountRole",
"AppUser",
"AssetClass",
"Broker",
"CashSnapshot",
"CashTxn",
"CashTxnTag",
"Category",
"CorporateAction",
"CorporateActionKind",
"CorporateActionStatus",
"Event",
"EventKind",
"EventSource",
"EventStatus",
"FlowType",
"FxRateDaily",
"Instrument",
"InstrumentAlias",
"JobStatus",
"Lot",
"LotDisposal",
"Merchant",
"MetricCashFlowMonthly",
"MetricDataQuality",
"MetricHolding",
"MetricNetWorthDaily",
"MetricPortfolioValueDaily",
"MetricRefreshLog",
"MetricReturns",
"MetricRunway",
"MetricSpendingByCategory",
"Portfolio",
"PortfolioAccount",
"PositionSnapshot",
"PriceDaily",
"PriceLast",
"PriceManual",
"RawCbrRate",
"RawTinvestEvent",
"RawTinvestInstrument",
"RawTinvestOperation",
"RawTinvestSnapshot",
"RawZenmoneyDeletion",
"RawZenmoneyEntity",
"RefreshToken",
"Rule",
"RuleKind",
"RuleMatchType",
"RunStatus",
"SourceCredential",
"SyncJob",
"SyncRun",
"SyncState",
"Trip",
]
+116
View File
@@ -0,0 +1,116 @@
"""Accounts, portfolios and the explicit ZenMoney <-> broker account mapping (plan §1.2)."""
from __future__ import annotations
import enum
from datetime import date, datetime
from decimal import Decimal
from typing import Any
from sqlalchemy import Boolean, ForeignKey, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from fintracker.db.base import Base, TimestampMixin, db_enum
class AccountKind(enum.StrEnum):
zm_cash = "zm_cash"
zm_card = "zm_card"
zm_checking = "zm_checking"
zm_deposit = "zm_deposit"
zm_loan = "zm_loan"
zm_emoney = "zm_emoney"
zm_debt = "zm_debt"
broker = "broker"
manual_asset = "manual_asset"
class AccountRole(enum.StrEnum):
liquid = "liquid"
savings = "savings"
investment = "investment"
debt = "debt"
class Broker(enum.StrEnum):
tinvest = "tinvest"
sber = "sber"
vtb = "vtb"
other = "other"
class EventSource(enum.StrEnum):
"""Which feed is the truth for an account's ledger; others become `shadow` events."""
tinvest_api = "tinvest_api"
report_sber = "report_sber"
report_vtb = "report_vtb"
manual = "manual"
class Account(TimestampMixin, Base):
__tablename__ = "account"
__table_args__ = (UniqueConstraint("source", "source_id"),)
id: Mapped[int] = mapped_column(primary_key=True)
kind: Mapped[AccountKind] = mapped_column(db_enum(AccountKind, "account_kind"))
source: Mapped[str] = mapped_column(String(32))
"""zenmoney | tinvest | report_sber | report_vtb | manual"""
source_id: Mapped[str] = mapped_column(String(128))
broker: Mapped[Broker | None] = mapped_column(db_enum(Broker, "broker"))
name: Mapped[str] = mapped_column(String(256))
currency: Mapped[str] = mapped_column(String(3))
include_in_net_worth: Mapped[bool] = mapped_column(Boolean, default=True)
role: Mapped[AccountRole] = mapped_column(db_enum(AccountRole, "account_role"))
mirror_of_account_id: Mapped[int | None] = mapped_column(
ForeignKey("account.id", ondelete="SET NULL")
)
"""ZenMoney account that merely mirrors a broker account (excluded from net worth)."""
primary_event_source: Mapped[EventSource | None] = mapped_column(
db_enum(EventSource, "event_source")
)
opened_at: Mapped[date | None]
archived: Mapped[bool] = mapped_column(Boolean, default=False)
deposit_terms: Mapped[dict[str, Any] | None]
"""ZenMoney deposit/loan terms: percent, startDate, endDateOffset, capitalization…"""
balance: Mapped[Decimal | None]
"""Current balance as the source reports it (ZenMoney `balance`); native currency."""
start_balance: Mapped[Decimal | None]
credit_limit: Mapped[Decimal | None]
balance_as_of: Mapped[datetime | None]
"""When `balance` was observed (set by the source sync). Informational only: the
net-worth walk in `analytics/networth.py` treats `balance` as the balance of TODAY
(`today_local()`) and steps back by transaction DATES, not by this timestamp."""
class Portfolio(TimestampMixin, Base):
__tablename__ = "portfolio"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(128), unique=True)
is_default: Mapped[bool] = mapped_column(Boolean, default=False)
base_currency: Mapped[str] = mapped_column(String(3), default="RUB")
class PortfolioAccount(Base):
__tablename__ = "portfolio_account"
portfolio_id: Mapped[int] = mapped_column(
ForeignKey("portfolio.id", ondelete="CASCADE"), primary_key=True
)
account_id: Mapped[int] = mapped_column(
ForeignKey("account.id", ondelete="CASCADE"), primary_key=True
)
class AccountLink(Base):
"""Explicit map used by the ZenMoney-transfer <-> broker-deposit matcher."""
__tablename__ = "account_link"
zm_account_id: Mapped[int] = mapped_column(
ForeignKey("account.id", ondelete="CASCADE"), primary_key=True
)
broker_account_id: Mapped[int] = mapped_column(
ForeignKey("account.id", ondelete="CASCADE"), unique=True
)
+33
View File
@@ -0,0 +1,33 @@
from __future__ import annotations
import uuid
from datetime import datetime
from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import Mapped, mapped_column
from fintracker.db.base import Base
class AppUser(Base):
"""The single user of this deployment. Kept as a table so a second one is not a rewrite."""
__tablename__ = "app_user"
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(String(320), unique=True)
password_hash: Mapped[str] = mapped_column(String(512))
created_at: Mapped[datetime] = mapped_column(server_default=func.now())
class RefreshToken(Base):
"""Refresh tokens are stored hashed so logout/revocation is real."""
__tablename__ = "refresh_token"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
user_id: Mapped[int] = mapped_column(ForeignKey("app_user.id", ondelete="CASCADE"), index=True)
token_hash: Mapped[str] = mapped_column(String(64), unique=True)
expires_at: Mapped[datetime]
revoked_at: Mapped[datetime | None]
created_at: Mapped[datetime] = mapped_column(server_default=func.now())
@@ -0,0 +1,85 @@
"""Instrument master (plan §1.3).
Identity resolution order: ISIN -> FIGI -> tinvest_uid -> (ticker, board) -> alias.
"""
from __future__ import annotations
import enum
from datetime import date
from decimal import Decimal
from typing import Any
from sqlalchemy import Boolean, ForeignKey, Index, Integer, String, UniqueConstraint, text
from sqlalchemy.orm import Mapped, mapped_column
from fintracker.db.base import Base, TimestampMixin, db_enum
class AssetClass(enum.StrEnum):
share = "share"
bond = "bond"
etf = "etf"
fund = "fund"
currency = "currency"
market_index = "index"
deposit = "deposit"
real_estate = "real_estate"
crypto = "crypto"
custom = "custom"
class Instrument(TimestampMixin, Base):
__tablename__ = "instrument"
__table_args__ = (
Index("uq_instrument_isin", "isin", unique=True, postgresql_where=text("isin IS NOT NULL")),
Index("uq_instrument_figi", "figi", unique=True, postgresql_where=text("figi IS NOT NULL")),
Index(
"uq_instrument_tinvest_uid",
"tinvest_uid",
unique=True,
postgresql_where=text("tinvest_uid IS NOT NULL"),
),
Index(
"uq_instrument_ticker_board",
"ticker",
"board",
unique=True,
postgresql_where=text("ticker IS NOT NULL AND board IS NOT NULL"),
),
)
id: Mapped[int] = mapped_column(primary_key=True)
asset_class: Mapped[AssetClass] = mapped_column(db_enum(AssetClass, "asset_class"))
isin: Mapped[str | None] = mapped_column(String(12))
figi: Mapped[str | None] = mapped_column(String(12))
tinvest_uid: Mapped[str | None] = mapped_column(String(64))
ticker: Mapped[str | None] = mapped_column(String(32))
board: Mapped[str | None] = mapped_column(String(16))
"""MOEX board / T-Invest class_code, e.g. TQBR, TQOB."""
exchange: Mapped[str | None] = mapped_column(String(32))
name: Mapped[str] = mapped_column(String(256))
issuer: Mapped[str | None] = mapped_column(String(256))
currency: Mapped[str] = mapped_column(String(3))
lot: Mapped[int] = mapped_column(Integer, default=1)
nominal: Mapped[Decimal | None]
nominal_currency: Mapped[str | None] = mapped_column(String(3))
maturity_date: Mapped[date | None]
sector: Mapped[str | None] = mapped_column(String(64))
country: Mapped[str | None] = mapped_column(String(2))
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
meta: Mapped[dict[str, Any] | None]
class InstrumentAlias(Base):
"""Source-specific keys resolving to an instrument, e.g. ('report_vtb', 'NAME:Газпром ао')."""
__tablename__ = "instrument_alias"
__table_args__ = (UniqueConstraint("source", "source_key"),)
id: Mapped[int] = mapped_column(primary_key=True)
instrument_id: Mapped[int] = mapped_column(
ForeignKey("instrument.id", ondelete="CASCADE"), index=True
)
source: Mapped[str] = mapped_column(String(32))
source_key: Mapped[str] = mapped_column(String(256))
+83
View File
@@ -0,0 +1,83 @@
"""Sync bookkeeping: cursors, run log, manual job queue, per-source credentials."""
from __future__ import annotations
import enum
import uuid
from datetime import datetime
from typing import Any
from sqlalchemy import ForeignKey, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column
from fintracker.db.base import Base, db_enum
class RunStatus(enum.StrEnum):
running = "running"
ok = "ok"
error = "error"
class JobStatus(enum.StrEnum):
queued = "queued"
running = "running"
done = "done"
error = "error"
class SyncState(Base):
"""One row per source: where the incremental sync left off."""
__tablename__ = "sync_state"
source: Mapped[str] = mapped_column(String(64), primary_key=True)
cursor: Mapped[str | None] = mapped_column(Text)
last_run_at: Mapped[datetime | None]
last_success_at: Mapped[datetime | None]
class SyncRun(Base):
__tablename__ = "sync_run"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
source: Mapped[str] = mapped_column(String(64), index=True)
status: Mapped[RunStatus] = mapped_column(db_enum(RunStatus, "run_status"))
triggered_by: Mapped[str] = mapped_column(String(32))
"""schedule | manual | cli"""
started_at: Mapped[datetime] = mapped_column(server_default=func.now())
finished_at: Mapped[datetime | None]
cursor_before: Mapped[str | None] = mapped_column(Text)
cursor_after: Mapped[str | None] = mapped_column(Text)
counts: Mapped[dict[str, Any] | None]
warnings: Mapped[list[Any] | None]
error: Mapped[str | None] = mapped_column(Text)
class SyncJob(Base):
"""Manual trigger queue: API inserts, worker picks up (plan §2.4)."""
__tablename__ = "sync_job"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
source: Mapped[str] = mapped_column(String(64), index=True)
status: Mapped[JobStatus] = mapped_column(db_enum(JobStatus, "job_status"), index=True)
requested_at: Mapped[datetime] = mapped_column(server_default=func.now())
started_at: Mapped[datetime | None]
finished_at: Mapped[datetime | None]
run_id: Mapped[uuid.UUID | None] = mapped_column(ForeignKey("sync_run.id", ondelete="SET NULL"))
error: Mapped[str | None] = mapped_column(Text)
class SourceCredential(Base):
"""Rotating credentials a source manages itself (e.g. ZenMoney refresh token).
Static tokens still come from the environment; this is for what the worker must
be able to update unattended.
"""
__tablename__ = "source_credential"
source: Mapped[str] = mapped_column(String(64), primary_key=True)
payload: Mapped[dict[str, Any]]
updated_at: Mapped[datetime] = mapped_column(server_default=func.now(), onupdate=func.now())
+1965
View File
File diff suppressed because it is too large Load Diff