statistics

Instructions

Implementation Reference

• src/mcp_cbr_rates/tools.py:254-301 (handler)

  Core implementation of the 'statistics' tool. Fetches USD/EUR/CNY rates, key rate, and inflation (year-over-year CPI) and returns a MacroSnapshot. Each sub-fetch is wrapped in a try/except so partial failures are tolerated (returns None for failed components).

  async def statistics(ctx: ToolContext) -> MacroSnapshot:
      """Return a compact macro snapshot: latest key rate, USD/EUR/CNY, inflation."""
      today = date.today()
  
      async def _safe_get_rate(code: str) -> Decimal | None:
          try:
              quote = await get_rate(ctx, code)
              return quote.vunit_rate
          except Exception as exc:
              logger.warning("statistics: cannot fetch %s rate: %s", code, exc)
              return None
  
      async def _safe_key_rate() -> Decimal | None:
          try:
              history = await key_rate(ctx)
              if history.points:
                  return history.points[-1].rate
              return None
          except Exception as exc:
              logger.warning("statistics: cannot fetch key rate: %s", exc)
              return None
  
      async def _safe_inflation() -> tuple[Decimal | None, str | None]:
          try:
              data = await inflation(ctx, year_from=today.year - 1, year_to=today.year)
              if data.points:
                  last = data.points[-1]
                  return last.cpi_yoy_pct, f"{last.year:04d}-{last.month:02d}"
              return None, None
          except Exception as exc:
              logger.warning("statistics: cannot fetch inflation: %s", exc)
              return None, None
  
      usd = await _safe_get_rate("USD")
      eur = await _safe_get_rate("EUR")
      cny = await _safe_get_rate("CNY")
      kr = await _safe_key_rate()
      infl_pct, infl_period = await _safe_inflation()
  
      return MacroSnapshot(
          as_of=today,
          key_rate_pct=kr,
          usd_rate=usd,
          eur_rate=eur,
          cny_rate=cny,
          inflation_yoy_pct=infl_pct,
          inflation_period=infl_period,
      )

• src/mcp_cbr_rates/schemas.py:104-118 (schema)

  MacroSnapshot Pydantic model: defines the return type for the statistics tool, with fields for as_of date, key_rate_pct, usd_rate, eur_rate, cny_rate, inflation_yoy_pct, and inflation_period.

  class MacroSnapshot(CbrModel):
      """High-level dashboard combining several CBR indicators."""
  
      as_of: _dt.date
      key_rate_pct: Decimal | None = None
      usd_rate: Decimal | None = None
      eur_rate: Decimal | None = None
      cny_rate: Decimal | None = None
      inflation_yoy_pct: Decimal | None = None
      inflation_period: str | None = Field(
          default=None,
          description="ISO 'YYYY-MM' string of the inflation observation included in the snapshot.",
      )
      source: str = DEFAULT_SOURCE

• src/mcp_cbr_rates/server.py:248-259 (registration)

  Registration of the 'statistics' tool with FastMCP: decorated with @mcp.tool(name='statistics', description=...). The handler function tool_statistics wraps the core implementation with error handling.

  @mcp.tool(
      name="statistics",
      description=(
          "Get a compact macro snapshot: latest key rate, USD/EUR/CNY rates,"
          " latest YoY inflation, and the period the inflation refers to."
      ),
  )
  async def tool_statistics(ctx: Context) -> MacroSnapshot:
      try:
          return await _statistics(_ctx(ctx))
      except CbrError as exc:
          raise RuntimeError(_format_error(exc)) from exc

• src/mcp_cbr_rates/server.py:52-54 (registration)

  Import of the statistics function from tools module (aliased as _statistics) used by the server's tool registration.

  from .tools import (
      statistics as _statistics,
  )

• tests/test_statistics.py:1-81 (helper)

  Tests for the statistics tool: verifies it combines all sources correctly, tolerates failed components, and handles missing currencies gracefully.

  """Tests for ``tools.statistics`` snapshot tool."""
  
  from __future__ import annotations
  
  from decimal import Decimal
  
  import httpx
  import pytest
  import respx
  
  from mcp_cbr_rates.tools import statistics
  
  from .conftest import load_fixture
  
  
  @pytest.mark.asyncio
  async def test_statistics_combines_all_sources(tool_ctx) -> None:
      daily = load_fixture("xml_daily_2024-04-25.xml")
      soap = load_fixture("soap_keyrate.xml")
      html = load_fixture("inflation.html")
      with respx.mock(base_url="https://www.cbr.ru") as router:
          router.get("/scripts/XML_daily.asp").mock(
              return_value=httpx.Response(200, content=daily)
          )
          router.post("/DailyInfoWebServ/DailyInfo.asmx").mock(
              return_value=httpx.Response(200, content=soap)
          )
          router.get("/hd_base/infl/").mock(
              return_value=httpx.Response(200, content=html)
          )
          snap = await statistics(tool_ctx)
      assert snap.usd_rate == Decimal("92.5012")
      assert snap.eur_rate == Decimal("99.1331")
      assert snap.cny_rate == Decimal("12.7843")
      assert snap.key_rate_pct == Decimal("16.00")
      assert snap.inflation_yoy_pct is not None
      assert snap.inflation_period is not None
  
  
  @pytest.mark.asyncio
  async def test_statistics_tolerates_failed_components(tool_ctx) -> None:
      daily = load_fixture("xml_daily_2024-04-25.xml")
      with respx.mock(base_url="https://www.cbr.ru") as router:
          router.get("/scripts/XML_daily.asp").mock(
              return_value=httpx.Response(200, content=daily)
          )
          router.post("/DailyInfoWebServ/DailyInfo.asmx").mock(
              return_value=httpx.Response(500, content=b"down")
          )
          router.get("/hd_base/infl/").mock(
              return_value=httpx.Response(500, content=b"down")
          )
          snap = await statistics(tool_ctx)
      assert snap.usd_rate == Decimal("92.5012")
      assert snap.key_rate_pct is None
      assert snap.inflation_yoy_pct is None
  
  
  @pytest.mark.asyncio
  async def test_statistics_handles_missing_currencies(tool_ctx) -> None:
      soap = load_fixture("soap_keyrate.xml")
      html = load_fixture("inflation.html")
      minimal_daily = (
          b'<?xml version="1.0" encoding="windows-1251"?>'
          b'<ValCurs Date="25.04.2024" name="Foreign Currency Market"></ValCurs>'
      )
      with respx.mock(base_url="https://www.cbr.ru") as router:
          router.get("/scripts/XML_daily.asp").mock(
              return_value=httpx.Response(200, content=minimal_daily)
          )
          router.post("/DailyInfoWebServ/DailyInfo.asmx").mock(
              return_value=httpx.Response(200, content=soap)
          )
          router.get("/hd_base/infl/").mock(
              return_value=httpx.Response(200, content=html)
          )
          snap = await statistics(tool_ctx)
      assert snap.usd_rate is None
      assert snap.eur_rate is None
      assert snap.key_rate_pct == Decimal("16.00")