feat(analytics): брокерские потоки по месяцам и ручные цены инструментов

GET /analytics/cashflow-broker читает metric_cash_flow_broker. POST
/instruments/{id}/prices апсертит price_manual по (instrument_id, d);
pricing/prices.py подмешивает его в ту же серию, что price_daily, с тем же
протягиванием и порогом устаревания.

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

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,
     )
 
 

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):

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)

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]
 ):