diff --git a/backend/src/fintracker/sources/tinvest/client.py b/backend/src/fintracker/sources/tinvest/client.py index bd502f3..cad78e2 100644 --- a/backend/src/fintracker/sources/tinvest/client.py +++ b/backend/src/fintracker/sources/tinvest/client.py @@ -30,6 +30,7 @@ from google.protobuf.json_format import MessageToDict from grpc import StatusCode from t_tech.invest import ( AsyncClient, + CandleInterval, GetOperationsByCursorRequest, InstrumentStatus, OperationItem, @@ -155,6 +156,19 @@ class InstrumentInfo: payload: dict[str, Any] +@dataclass(frozen=True) +class DayCandle: + """One finished daily candle. For a bond `close` is percent of nominal, as the broker + quotes it; for everything else it is the price in the instrument's currency.""" + + d: date + close: Decimal + open: Decimal | None + high: Decimal | None + low: Decimal | None + volume: Decimal | None + + @dataclass(frozen=True) class DividendRow: """One `Dividend` from GetDividends, flattened. @@ -491,6 +505,47 @@ class TinvestClient: country=info.country or match.country, ) + async def daily_candles( + self, instrument_uid: str, *, since: date, until: date + ) -> list[DayCandle]: + """Finished daily candles over [since, until] — the history of a paper MOEX has none of. + + The day still forming is left out: it is not a close yet, and today's price is already + the broker's mark. NOT_FOUND (a paper the market-data service does not serve) is an + empty list, not a failure of the whole sync. + """ + start = datetime(since.year, since.month, since.day, tzinfo=UTC) + end = datetime(until.year, until.month, until.day, 23, 59, 59, tzinfo=UTC) + try: + resp = await self._call( + lambda: self._client.market_data.get_candles( + instrument_id=instrument_uid, + from_=start, + to=end, + interval=CandleInterval.CANDLE_INTERVAL_DAY, + ) + ) + except AioRequestError as err: + if err.code is StatusCode.NOT_FOUND: + return [] + raise + out: list[DayCandle] = [] + for c in resp.candles: + close = quotation_to_decimal(c.close) + if not c.is_complete or close is None: + continue + out.append( + DayCandle( + d=c.time.astimezone(UTC).date(), + close=close, + open=quotation_to_decimal(c.open), + high=quotation_to_decimal(c.high), + low=quotation_to_decimal(c.low), + volume=Decimal(c.volume), + ) + ) + return out + async def dividends( self, instrument_uid: str, *, since: datetime, until: datetime ) -> list[DividendRow]: diff --git a/backend/src/fintracker/sources/tinvest/sync.py b/backend/src/fintracker/sources/tinvest/sync.py index 8174488..42648c9 100644 --- a/backend/src/fintracker/sources/tinvest/sync.py +++ b/backend/src/fintracker/sources/tinvest/sync.py @@ -8,7 +8,8 @@ Flow of one run: operation id), then mapped into `event`. 3. Instruments seen in those operations are resolved once and stored in `instrument`. 4. `GetPortfolio`/`GetPositions` -> snapshots, for the derived-vs-reported check; for a paper - MOEX never quotes, the broker's own mark also becomes that day's price (`_price_from_broker`). + MOEX never quotes, the broker's own mark also becomes that day's price (`_price_from_broker`), + and its daily candles give the history behind it (`_backfill_broker_prices`). **Cursor.** T-Invest's own cursor is a per-request token, not a durable watermark, so it cannot be stored between runs. Instead the cursor is a JSON map `{account_id: iso_ts}` of @@ -30,7 +31,7 @@ from decimal import Decimal from typing import Any from zoneinfo import ZoneInfo -from sqlalchemy import delete, func, select +from sqlalchemy import delete, func, select, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession @@ -105,6 +106,7 @@ class TinvestSource: "instruments": 0, "sectors": 0, "snapshots": 0, + "prices": 0, } warnings: list[str] = [] new_cursors: dict[str, str] = dict(cursors) @@ -144,6 +146,7 @@ class TinvestSource: ) counts["sectors"] = await _backfill_sectors(session, client) + counts["prices"] = await _backfill_broker_prices(session, client) await _fill_logos(session) @@ -804,3 +807,130 @@ async def _price_from_broker(session: AsyncSession, snapshot: Any, by_uid: dict[ ) ) return len(rows) + + +BROKER_HISTORY_DAYS = 365 +"""How far back to ask when a paper has no purchase in the ledger to start from.""" +BROKER_HISTORY_MARGIN = timedelta(days=7) +BROKER_TAIL_OVERLAP = timedelta(days=5) + + +async def _backfill_broker_prices(session: AsyncSession, client: TinvestClient) -> int: + """The price history of the papers only the broker quotes, from its daily candles. + + `_price_from_broker` records today's mark and nothing before it, so such a paper's chart + holds one point and its value is unknown for every day since the purchase. The candles + fill that in: the first run reaches back to a week before the first purchase, later runs + re-read only the last few days. + + Bonds arrive in percent of nominal and become money the way `sources/moex` does it + (`price_pct / 100 * nominal`); a nominal in a foreign currency (a yuan bond) is taken to + roubles at that day's CBR rate, the currency the broker's own mark is in — otherwise the + series would jump between two currencies on the day the mark begins. A day with no rate is + skipped, never guessed. Rows another source wrote are left alone. + """ + from fintracker.analytics import today_local + from fintracker.pricing.fx import FxTable + + moex_priced = select(PriceDaily.instrument_id).where(PriceDaily.source == "moex") + marked = ( + await session.execute( + select( + Instrument.id, + Instrument.tinvest_uid, + Instrument.asset_class, + Instrument.currency, + Instrument.nominal, + Instrument.nominal_currency, + ).where( + Instrument.tinvest_uid.is_not(None), + Instrument.id.in_( + select(PriceDaily.instrument_id).where(PriceDaily.source == SOURCE) + ), + Instrument.id.not_in(moex_priced), + ) + ) + ).all() + if not marked: + return 0 + + today = today_local() + fx = await FxTable.load(session) + stored = 0 + for inst in marked: + first_buy = await session.scalar( + select(func.min(Event.trade_date)).where(Event.instrument_id == inst.id) + ) + wanted = (first_buy or today - timedelta(days=BROKER_HISTORY_DAYS)) - BROKER_HISTORY_MARGIN + have_from, have_to = ( + await session.execute( + select(func.min(PriceDaily.d), func.max(PriceDaily.d)).where( + PriceDaily.instrument_id == inst.id, PriceDaily.source == SOURCE + ) + ) + ).one() + # a history that already reaches the wanted start only needs its tail refreshed + since = ( + wanted + if have_from is None or have_from > wanted + BROKER_HISTORY_MARGIN + else ((have_to or today) - BROKER_TAIL_OVERLAP) + ) + is_bond = inst.asset_class == AssetClass.bond + nominal, nominal_ccy = inst.nominal, inst.nominal_currency + if is_bond and nominal is None: + # GetInstrumentBy does not state the nominal; the per-type bond listing does + listed = await client.reference_instrument("bond", uid=inst.tinvest_uid) + if listed is not None and listed.nominal is not None: + nominal, nominal_ccy = listed.nominal, listed.nominal_currency + await session.execute( + update(Instrument) + .where(Instrument.id == inst.id) + .values(nominal=nominal, nominal_currency=nominal_ccy) + ) + candles = await client.daily_candles(inst.tinvest_uid, since=since, until=today) + currency = (nominal_ccy or inst.currency or "RUB").upper() if is_bond else None + rows = [] + for c in candles: + if is_bond: + if nominal is None: + break + money = c.close / Decimal(100) * nominal + close = fx.to_rub(money, currency, c.d) + if close is None: + continue + row_currency, pct = "RUB", c.close + else: + close, row_currency, pct = c.close, (inst.currency or "RUB").upper(), None + rows.append( + { + "instrument_id": inst.id, + "d": c.d, + "close": close, + "open": None, + "high": None, + "low": None, + "volume": c.volume, + "currency": row_currency, + "source": SOURCE, + "price_pct": pct, + } + ) + if not rows: + continue + stmt = pg_insert(PriceDaily).values(rows) + await session.execute( + stmt.on_conflict_do_update( + index_elements=["instrument_id", "d"], + set_={ + "close": stmt.excluded.close, + "volume": stmt.excluded.volume, + "currency": stmt.excluded.currency, + "price_pct": stmt.excluded.price_pct, + }, + where=PriceDaily.source == SOURCE, + ) + ) + stored += len(rows) + if stored: + log.info("tinvest: %s broker-quoted daily prices from candles", stored) + return stored diff --git a/backend/tests/sources/test_tinvest_candles.py b/backend/tests/sources/test_tinvest_candles.py new file mode 100644 index 0000000..a29375e --- /dev/null +++ b/backend/tests/sources/test_tinvest_candles.py @@ -0,0 +1,208 @@ +"""A paper only the broker quotes gets its price history from the broker's daily candles.""" + +from __future__ import annotations + +from datetime import date, timedelta +from decimal import Decimal + +from sqlalchemy import select + +from factories import make_account, make_event, make_instrument +from fintracker.db import get_sessionmaker +from fintracker.models import ( + AccountKind, + AccountRole, + AssetClass, + EventKind, + FxRateDaily, + Instrument, +) +from fintracker.models.pricing import PriceDaily +from fintracker.sources.tinvest.client import DayCandle, InstrumentInfo +from fintracker.sources.tinvest.sync import _backfill_broker_prices + +D = Decimal +TODAY = date(2026, 9, 19) + + +class FakeClient: + """Only what `_backfill_broker_prices` calls; remembers what it was asked for.""" + + def __init__(self, candles: list[DayCandle], *, nominal: InstrumentInfo | None = None) -> None: + self.candles = candles + self.nominal = nominal + self.listed = 0 + self.asked: list[tuple[str, date, date]] = [] + + async def reference_instrument(self, kind: str, *, uid: str) -> InstrumentInfo | None: + self.listed += 1 + return self.nominal + + async def daily_candles(self, uid: str, *, since: date, until: date) -> list[DayCandle]: + self.asked.append((uid, since, until)) + return [c for c in self.candles if since <= c.d <= until] + + +def candle(d: date, close: str) -> DayCandle: + return DayCandle(d=d, close=D(close), open=None, high=None, low=None, volume=D(10)) + + +async def bond(*, nominal_currency: str = "CNY") -> int: + """A yuan bond held since 2026-04-03, marked by the broker today in roubles.""" + iid = await make_instrument(ticker="SIBN6P4", asset_class=AssetClass.bond, board=None) + account = await make_account( + name="Т", kind=AccountKind.broker, role=AccountRole.investment, balance=None + ) + await make_event( + date(2026, 4, 3), account_id=account, kind=EventKind.buy, instrument_id=iid, + quantity=1, price=12000, amount=-12000, + ) # fmt: skip + async with get_sessionmaker()() as session: + row = await session.get(Instrument, iid) + assert row is not None + row.tinvest_uid, row.nominal, row.nominal_currency = "uid", D(1000), nominal_currency + session.add( + PriceDaily( + instrument_id=iid, d=TODAY, close=D("12354.6177"), currency="RUB", source="tinvest" + ) + ) + await session.commit() + return iid + + +async def rates(ccy: str, value: str, *days: date) -> None: + async with get_sessionmaker()() as session: + session.add_all( + FxRateDaily(d=d, ccy=ccy, rate_rub=D(value), is_carried=False) for d in days + ) + await session.commit() + + +async def stored(iid: int) -> dict[date, PriceDaily]: + async with get_sessionmaker()() as session: + rows = ( + await session.execute(select(PriceDaily).where(PriceDaily.instrument_id == iid)) + ).scalars() + return {r.d: r for r in rows} + + +async def run(client: FakeClient) -> int: + async with get_sessionmaker()() as session: + # the sync passes `TODAY` through `today_local()`; the fake ignores the upper bound + n = await _backfill_broker_prices(session, client) # type: ignore[arg-type] + await session.commit() + return n + + +async def test_a_yuan_bond_gets_its_history_in_roubles(app, monkeypatch): + monkeypatch.setattr("fintracker.analytics.today_local", lambda: TODAY) + iid = await bond() + await rates("CNY", "12.5", date(2026, 9, 17), date(2026, 9, 18)) + client = FakeClient([candle(date(2026, 9, 17), "98.4"), candle(date(2026, 9, 18), "98.31")]) + + assert await run(client) == 2 + + rows = await stored(iid) + # 98.31 % of 1000 CNY at 12.5 ₽ — the mark is in roubles, so the history is too + assert rows[date(2026, 9, 18)].close == D("12288.75") + assert (rows[date(2026, 9, 18)].currency, rows[date(2026, 9, 18)].price_pct) == ( + "RUB", + D("98.31"), + ) + assert rows[TODAY].close == D("12354.6177") # the broker's mark is not touched + # first run reaches back to a week before the purchase + assert client.asked[0][1] == date(2026, 4, 3) - timedelta(days=7) + + +async def test_a_day_without_a_rate_is_skipped_not_guessed(app, monkeypatch): + monkeypatch.setattr("fintracker.analytics.today_local", lambda: TODAY) + iid = await bond() + await rates("CNY", "12.5", date(2026, 9, 18)) + client = FakeClient([candle(date(2026, 9, 17), "98.4"), candle(date(2026, 9, 18), "98.31")]) + + assert await run(client) == 1 + assert set(await stored(iid)) == {date(2026, 9, 18), TODAY} + + +async def test_a_second_run_reads_only_the_tail(app, monkeypatch): + monkeypatch.setattr("fintracker.analytics.today_local", lambda: TODAY) + await bond() + await rates("CNY", "12.5", *(date(2026, 4, 1) + timedelta(days=n) for n in range(200))) + first = FakeClient([candle(date(2026, 4, 1) + timedelta(days=n), "99") for n in range(170)]) + await run(first) + + second = FakeClient([]) + await run(second) + + assert second.asked[0][1] > date(2026, 9, 1) + + +async def test_a_paper_the_exchange_prices_is_left_to_the_exchange(app, monkeypatch): + monkeypatch.setattr("fintracker.analytics.today_local", lambda: TODAY) + iid = await bond() + async with get_sessionmaker()() as session: + session.add( + PriceDaily( + instrument_id=iid, d=date(2026, 9, 1), close=D(1), currency="RUB", source="moex" + ) + ) + await session.commit() + client = FakeClient([candle(date(2026, 9, 18), "98")]) + + assert await run(client) == 0 + assert client.asked == [] + + +async def test_a_share_keeps_its_own_currency_and_price(app, monkeypatch): + monkeypatch.setattr("fintracker.analytics.today_local", lambda: TODAY) + iid = await make_instrument(ticker="AAPL", currency="USD", board=None) + async with get_sessionmaker()() as session: + row = await session.get(Instrument, iid) + assert row is not None + row.tinvest_uid = "uid-aapl" + session.add( + PriceDaily(instrument_id=iid, d=TODAY, close=D(200), currency="USD", source="tinvest") + ) + await session.commit() + + assert await run(FakeClient([candle(date(2026, 9, 18), "199.5")])) == 1 + row = (await stored(iid))[date(2026, 9, 18)] + assert (row.close, row.currency, row.price_pct) == (D("199.5"), "USD", None) + + +async def test_a_bond_with_no_nominal_asks_the_bond_listing_and_remembers_it(app, monkeypatch): + monkeypatch.setattr("fintracker.analytics.today_local", lambda: TODAY) + iid = await bond() + async with get_sessionmaker()() as session: + row = await session.get(Instrument, iid) + assert row is not None + row.nominal = row.nominal_currency = None + await session.commit() + await rates("CNY", "12.5", date(2026, 9, 18)) + listed = InstrumentInfo( + uid="uid", kind="bond", isin=None, figi=None, ticker="SIBN6P4", class_code=None, + name="x", currency="rub", lot=1, nominal=D(1000), nominal_currency="CNY", + maturity_date=None, sector=None, country=None, exchange=None, payload={}, + ) # fmt: skip + client = FakeClient([candle(date(2026, 9, 18), "98.31")], nominal=listed) + + assert await run(client) == 1 + + assert (await stored(iid))[date(2026, 9, 18)].close == D("12288.75") + async with get_sessionmaker()() as session: + row = await session.get(Instrument, iid) + assert row is not None + assert (row.nominal, row.nominal_currency) == (D(1000), "CNY") + + +async def test_a_bond_the_listing_does_not_know_stays_unpriced_by_candles(app, monkeypatch): + monkeypatch.setattr("fintracker.analytics.today_local", lambda: TODAY) + iid = await bond() + async with get_sessionmaker()() as session: + row = await session.get(Instrument, iid) + assert row is not None + row.nominal = None + await session.commit() + + assert await run(FakeClient([candle(date(2026, 9, 18), "98.31")])) == 0 + assert set(await stored(iid)) == {TODAY}