feat(tinvest): история цен бумаги, которую MOEX не котирует, по дневным свечам брокера

TinvestClient.daily_candles отдаёт завершённые дневные свечи, sync добирает ими историю к цене брокера (_backfill_broker_prices). Тесты БД в этой среде не запускались (нет pg_config).
2026-09-19 22:13:42 +03:00
parent 29bf512735
commit 75933993b2
3 changed files with 395 additions and 2 deletions
@@ -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]:
@@ -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