Dmitry
b9c12fa1a1
fx_rate_daily получает строку на каждый календарный день: котировки ЦБ тянутся вперёд (и назад до первой), is_carried это помечает, RUB = 1.0 всегда. Дальше любая сумма конвертируется по курсу СВОЕЙ даты, а не сегодняшнему. Net worth восстанавливается назад от текущего account.balance по транзакциям — ZenMoney отдаёт остаток, а не историю; поэтому сегодняшняя строка совпадает с тем, что показывает ZenMoney, а каждая прошлая с ней согласована. Нет курса — не подстановка, а NULL и строка в metric_data_quality. Туда же попадает то, что шаги заметили по дороге: правило без совпадений, счёт без баланса, перевод через границу net worth.
114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy import select
|
|
|
|
from factories import make_cbr_rate, make_txn
|
|
from fintracker.analytics import today_local
|
|
from fintracker.db import get_sessionmaker
|
|
from fintracker.models import FxRateDaily
|
|
from fintracker.pricing.fx import FxTable, rebuild_fx_daily
|
|
|
|
|
|
def last_friday(before_days: int = 7) -> date:
|
|
d = today_local() - timedelta(days=before_days)
|
|
return d - timedelta(days=(d.weekday() - 4) % 7)
|
|
|
|
|
|
async def rebuild() -> None:
|
|
async with get_sessionmaker()() as session:
|
|
await rebuild_fx_daily(session)
|
|
await session.commit()
|
|
|
|
|
|
async def rates_on(d: date) -> dict[str, tuple[Decimal, bool]]:
|
|
async with get_sessionmaker()() as session:
|
|
rows = (await session.execute(select(FxRateDaily).where(FxRateDaily.d == d))).scalars()
|
|
return {r.ccy: (r.rate_rub, r.is_carried) for r in rows}
|
|
|
|
|
|
async def test_nominal_is_divided_out(app):
|
|
friday = last_friday()
|
|
await make_cbr_rate(friday, "JPY", "65.0", nominal=100)
|
|
await rebuild()
|
|
|
|
rate, is_carried = (await rates_on(friday))["JPY"]
|
|
assert rate == Decimal("0.65")
|
|
assert is_carried is False
|
|
|
|
|
|
async def test_weekend_carries_friday_forward(app):
|
|
friday = last_friday()
|
|
await make_cbr_rate(friday, "USD", "90.5")
|
|
await rebuild()
|
|
|
|
for offset in (1, 2): # Saturday, Sunday
|
|
rate, is_carried = (await rates_on(friday + timedelta(days=offset)))["USD"]
|
|
assert rate == Decimal("90.5")
|
|
assert is_carried is True
|
|
|
|
|
|
async def test_days_before_the_first_quote_are_back_filled(app):
|
|
friday = last_friday()
|
|
earlier = friday - timedelta(days=10)
|
|
await make_txn(earlier, outcome="100", outcome_currency="USD")
|
|
await make_cbr_rate(friday, "USD", "90.5")
|
|
await rebuild()
|
|
|
|
rate, is_carried = (await rates_on(earlier))["USD"]
|
|
assert rate == Decimal("90.5")
|
|
assert is_carried is True
|
|
|
|
|
|
async def test_rub_is_one_on_every_day_and_outside_the_spine(app):
|
|
friday = last_friday()
|
|
await make_cbr_rate(friday, "USD", "90.5")
|
|
await rebuild()
|
|
|
|
assert (await rates_on(friday))["RUB"] == (Decimal(1), False)
|
|
assert (await rates_on(today_local()))["RUB"] == (Decimal(1), False)
|
|
|
|
async with get_sessionmaker()() as session:
|
|
fx = await FxTable.load(session)
|
|
assert fx.rate(date(1999, 1, 1), "RUB") == Decimal(1)
|
|
assert fx.rate(date(1999, 1, 1), "USD") is None
|
|
assert fx.to_rub(Decimal("10"), "USD", friday) == Decimal("905.0")
|
|
assert fx.to_rub(Decimal("10"), "XBT", friday) is None
|
|
|
|
|
|
async def test_spine_covers_future_rates_and_future_transactions(app):
|
|
"""The CBR publishes tomorrow's rate the evening before, and a transaction may be dated
|
|
in the future — both days must be convertible."""
|
|
t = today_local()
|
|
await make_cbr_rate(t, "USD", "90")
|
|
await make_cbr_rate(t + timedelta(days=1), "USD", "95")
|
|
await make_txn(t + timedelta(days=3), outcome="10", outcome_currency="USD")
|
|
await rebuild()
|
|
|
|
assert (await rates_on(t + timedelta(days=1)))["USD"] == (Decimal("95"), False)
|
|
# the txn is dated past the last quote: the spine still reaches it, carried forward
|
|
assert (await rates_on(t + timedelta(days=3)))["USD"] == (Decimal("95"), True)
|
|
|
|
async with get_sessionmaker()() as session:
|
|
fx = await FxTable.load(session)
|
|
assert fx.to_rub(Decimal("10"), "USD", t + timedelta(days=3)) == Decimal("950")
|
|
|
|
|
|
async def test_deleted_txn_does_not_stretch_the_spine(app):
|
|
"""ZenMoney hands out a zero date (1970-01-01) for some deleted rows.
|
|
|
|
Counting it would build the daily grid over five extra decades of carried-forward rates.
|
|
"""
|
|
friday = last_friday()
|
|
await make_cbr_rate(friday, "USD", "90.5")
|
|
await make_txn(friday, outcome=100, outcome_currency="RUB")
|
|
await make_txn(date(1970, 1, 1), income=15000, income_currency="RUB", deleted=True)
|
|
await rebuild()
|
|
|
|
async with get_sessionmaker()() as session:
|
|
earliest = (
|
|
(await session.execute(select(FxRateDaily.d).order_by(FxRateDaily.d))).scalars().first()
|
|
)
|
|
assert earliest is not None
|
|
assert earliest >= friday - timedelta(days=1)
|