From 6b32c79e79f1032fe2798daf0c19ec8bdc424288 Mon Sep 17 00:00:00 2001 From: Dmitry Date: Sat, 19 Sep 2026 10:38:59 +0300 Subject: [PATCH] =?UTF-8?q?feat(analytics):=20=D0=B1=D1=80=D0=BE=D0=BA?= =?UTF-8?q?=D0=B5=D1=80=D1=81=D0=BA=D0=B8=D0=B5=20=D0=BF=D0=BE=D1=82=D0=BE?= =?UTF-8?q?=D0=BA=D0=B8=20=D0=BF=D0=BE=20=D0=BC=D0=B5=D1=81=D1=8F=D1=86?= =?UTF-8?q?=D0=B0=D0=BC=20=D0=B8=20=D1=80=D1=83=D1=87=D0=BD=D1=8B=D0=B5=20?= =?UTF-8?q?=D1=86=D0=B5=D0=BD=D1=8B=20=D0=B8=D0=BD=D1=81=D1=82=D1=80=D1=83?= =?UTF-8?q?=D0=BC=D0=B5=D0=BD=D1=82=D0=BE=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GET /analytics/cashflow-broker читает metric_cash_flow_broker. POST /instruments/{id}/prices апсертит price_manual по (instrument_id, d); pricing/prices.py подмешивает его в ту же серию, что price_daily, с тем же протягиванием и порогом устаревания. --- .../src/fintracker/api/routers/analytics.py | 28 ++++++++ .../src/fintracker/api/routers/instruments.py | 72 +++++++++++++++++-- .../src/fintracker/api/schemas/analytics.py | 27 +++++++ backend/src/fintracker/pricing/prices.py | 24 ++++++- backend/tests/api/test_analytics_api.py | 62 ++++++++++++++++ 5 files changed, 205 insertions(+), 8 deletions(-) diff --git a/backend/src/fintracker/api/routers/analytics.py b/backend/src/fintracker/api/routers/analytics.py index 20d0edc..66d8c13 100644 --- a/backend/src/fintracker/api/routers/analytics.py +++ b/backend/src/fintracker/api/routers/analytics.py @@ -17,6 +17,7 @@ from sqlalchemy import func, select from fintracker.api.deps import CurrentUser, SessionDep from fintracker.api.schemas.analytics import ( AllocationBucket, + CashFlowBrokerMonth, HoldingOut, ReturnsOut, ScopeOut, @@ -28,6 +29,7 @@ from fintracker.models import ( AllocationDimension, Instrument, MetricAllocation, + MetricCashFlowBroker, MetricHolding, MetricPortfolioValueDaily, MetricRefreshLog, @@ -137,6 +139,32 @@ async def allocation( ] +@router.get("/cashflow-broker", name="cashflow_broker") +async def cashflow_broker( + session: SessionDep, _: CurrentUser, scope: ScopeParam = DEFAULT_SCOPE +) -> list[CashFlowBrokerMonth]: + """Money moved across the brokerage boundary, by month: the other half of cash flow + that `metric_cash_flow_monthly` (ZenMoney spending) does not cover.""" + await resolve_scope(session, scope) + rows = ( + await session.execute( + select(MetricCashFlowBroker) + .where(MetricCashFlowBroker.scope == scope) + .order_by(MetricCashFlowBroker.month) + ) + ).scalars() + return [ + CashFlowBrokerMonth( + month=r.month, + deposits_rub=r.deposits_rub, + withdrawals_rub=r.withdrawals_rub, + net_rub=r.net_rub, + event_count=r.event_count, + ) + for r in rows + ] + + @router.get("/summary", name="summary") async def summary( session: SessionDep, _: CurrentUser, scope: ScopeParam = DEFAULT_SCOPE diff --git a/backend/src/fintracker/api/routers/instruments.py b/backend/src/fintracker/api/routers/instruments.py index c5cd160..0407038 100644 --- a/backend/src/fintracker/api/routers/instruments.py +++ b/backend/src/fintracker/api/routers/instruments.py @@ -9,7 +9,7 @@ from __future__ import annotations from datetime import date, timedelta from typing import Annotated -from fastapi import APIRouter, Query +from fastapi import APIRouter, Query, status from sqlalchemy import select from fintracker.analytics import today_local @@ -21,10 +21,12 @@ from fintracker.api.schemas.analytics import ( InstrumentDetail, InstrumentOut, LotOut, + PriceManualIn, + PriceManualOut, PricePoint, ) from fintracker.api.scopes import DEFAULT_SCOPE, resolve_scope -from fintracker.models import AssetClass, Instrument, Lot, PriceDaily +from fintracker.models import AssetClass, Instrument, Lot, PriceDaily, PriceManual router = APIRouter(prefix="/instruments", tags=["instruments"]) @@ -103,6 +105,24 @@ async def get_instrument( .order_by(PriceDaily.d) ) ).scalars() + manual_prices = ( + await session.execute( + select(PriceManual) + .where(PriceManual.instrument_id == instrument_id, PriceManual.d >= since) + .order_by(PriceManual.d) + ) + ).scalars() + + points = [ + PricePoint(d=p.d, close=p.close, currency=p.currency, accrued_interest=p.accrued_interest) + for p in prices + ] + [ + PricePoint( + d=p.d, close=p.price, currency=p.currency, accrued_interest=None, source="manual" + ) + for p in manual_prices + ] + points.sort(key=lambda p: p.d) return InstrumentDetail( instrument=_instrument(instrument), @@ -122,12 +142,50 @@ async def get_instrument( for lot in lots ], events=await events_of_instrument(session, instrument_id), - prices=[ - PricePoint( - d=p.d, close=p.close, currency=p.currency, accrued_interest=p.accrued_interest + prices=points, + ) + + +@router.post( + "/{instrument_id}/prices", name="set_manual_price", status_code=status.HTTP_201_CREATED +) +async def set_manual_price( + instrument_id: int, body: PriceManualIn, session: SessionDep, _: CurrentUser +) -> PriceManualOut: + """Set (or replace) the manual price of `instrument_id` on `body.d`. + + For what no exchange quotes: real estate, crypto, a custom holding, an OTC bond nobody + lists. One row per (instrument, day) — a second call for the same day replaces it rather + than piling up duplicates the price series would then have to pick between. Does not + itself refresh `metric_*`; call `POST /metrics/refresh` after, same as `/rules/apply`. + """ + instrument = await session.get(Instrument, instrument_id) + if instrument is None: + raise Problem(404, "Not Found", f"Нет инструмента #{instrument_id}") + + existing = ( + await session.execute( + select(PriceManual).where( + PriceManual.instrument_id == instrument_id, PriceManual.d == body.d ) - for p in prices - ], + ) + ).scalar_one_or_none() + if existing is None: + existing = PriceManual(instrument_id=instrument_id, d=body.d) + session.add(existing) + existing.price = body.price + existing.currency = body.currency.upper() + existing.note = body.note + + await session.commit() + await session.refresh(existing) + return PriceManualOut( + id=existing.id, + instrument_id=existing.instrument_id, + d=existing.d, + price=existing.price, + currency=existing.currency, + note=existing.note, ) diff --git a/backend/src/fintracker/api/schemas/analytics.py b/backend/src/fintracker/api/schemas/analytics.py index 16415fe..060a857 100644 --- a/backend/src/fintracker/api/schemas/analytics.py +++ b/backend/src/fintracker/api/schemas/analytics.py @@ -166,6 +166,33 @@ class PricePoint(BaseModel): close: Money currency: str accrued_interest: MoneyOpt + source: str = "daily" + """"daily" (price_daily, from an exchange) or "manual" (`price_manual`).""" + + +class PriceManualIn(BaseModel): + d: date + price: Money + currency: str + note: str | None = None + + +class PriceManualOut(BaseModel): + id: int + instrument_id: int + d: date + price: Money + currency: str + note: str | None + + +class CashFlowBrokerMonth(BaseModel): + month: date + """First day of the month.""" + deposits_rub: Money + withdrawals_rub: Money + net_rub: Money + event_count: int class InstrumentOut(BaseModel): diff --git a/backend/src/fintracker/pricing/prices.py b/backend/src/fintracker/pricing/prices.py index dbdff59..c39d8f9 100644 --- a/backend/src/fintracker/pricing/prices.py +++ b/backend/src/fintracker/pricing/prices.py @@ -23,7 +23,7 @@ from decimal import Decimal from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from fintracker.models import PriceDaily, PriceLast +from fintracker.models import PriceDaily, PriceLast, PriceManual STALE_AFTER_DAYS = 10 """Beyond this, a carried-forward close is still used but reported as stale (plan §7.8).""" @@ -86,6 +86,28 @@ class PriceTable: ) ) + # price_manual fills gaps for what no exchange quotes (plan §1.6): a real-estate or + # OTC holding with no price_daily rows at all. Merged into the same series so `at`'s + # bisect and the stale-after-N-days rule apply to it exactly like an exchange close. + manual_rows = ( + await session.execute( + select( + PriceManual.instrument_id, + PriceManual.d, + PriceManual.price, + PriceManual.currency, + ).order_by(PriceManual.instrument_id, PriceManual.d) + ) + ).all() + touched: set[int] = set() + for r in manual_rows: + series.setdefault(r.instrument_id, []).append( + Quote(price=Decimal(r.price), currency=r.currency.upper(), as_of=r.d) + ) + touched.add(r.instrument_id) + for iid in touched: + series[iid].sort(key=lambda q: q.as_of) + last_rows = ( await session.execute( select(PriceLast.instrument_id, PriceLast.ts, PriceLast.price, PriceLast.currency) diff --git a/backend/tests/api/test_analytics_api.py b/backend/tests/api/test_analytics_api.py index 1884f80..3d4345c 100644 --- a/backend/tests/api/test_analytics_api.py +++ b/backend/tests/api/test_analytics_api.py @@ -160,6 +160,68 @@ async def test_the_instrument_card_carries_lots_events_and_prices( assert body["prices"][0]["close"] == "110.0000000000" +async def test_manual_price_fills_the_gap_and_shows_up_as_manual( + client: AsyncClient, auth_headers: dict[str, str], portfolio: dict[str, int] +): + silent = portfolio["silent"] + d = today_local().isoformat() + + r = await client.post( + f"/api/v1/instruments/{silent}/prices", + headers=auth_headers, + json={"d": d, "price": "1050", "currency": "RUB", "note": "ручная оценка"}, + ) + assert r.status_code == 201, r.text + assert r.json()["price"] == "1050.0000000000" + + r = await client.get(f"/api/v1/instruments/{silent}", headers=auth_headers) + prices = r.json()["prices"] + assert prices[-1] == { + "d": d, + "close": "1050.0000000000", + "currency": "RUB", + "accrued_interest": None, + "source": "manual", + } + + # a second call for the same day replaces it rather than adding a row + r = await client.post( + f"/api/v1/instruments/{silent}/prices", + headers=auth_headers, + json={"d": d, "price": "1100", "currency": "RUB"}, + ) + assert r.status_code == 201, r.text + r = await client.get(f"/api/v1/instruments/{silent}", headers=auth_headers) + manual_prices = [p for p in r.json()["prices"] if p["source"] == "manual"] + assert len(manual_prices) == 1 + assert manual_prices[0]["close"] == "1100.0000000000" + + +async def test_manual_price_on_an_unknown_instrument_is_a_problem( + client: AsyncClient, auth_headers: dict[str, str] +): + r = await client.post( + "/api/v1/instruments/999999/prices", + headers=auth_headers, + json={"d": today_local().isoformat(), "price": "1", "currency": "RUB"}, + ) + assert r.status_code == 404 + + +async def test_cashflow_broker_reports_the_months_money_moved( + client: AsyncClient, auth_headers: dict[str, str], portfolio: dict[str, int] +): + r = await client.get("/api/v1/analytics/cashflow-broker", headers=auth_headers) + assert r.status_code == 200, r.text + rows = r.json() + assert rows # the fixture's 20000 ₽ deposit lands in exactly one month + total_deposits = sum(float(row["deposits_rub"]) for row in rows) + total_withdrawals = sum(float(row["withdrawals_rub"]) for row in rows) + assert total_deposits == 20000.0 + assert total_withdrawals == 0.0 + assert all(row["net_rub"] == row["deposits_rub"] for row in rows) + + async def test_events_can_be_filtered_down_to_the_external_flows( client: AsyncClient, auth_headers: dict[str, str], portfolio: dict[str, int] ):