Adriano/Cerbero-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
1ca1687c9b325ff0d4af7106ce99fad6a091a603
Cerbero-mcp/tests/unit/exchanges/macro/test_fetchers.py
T
AdrianoDev 88bd4e7bde feat(V2): migrazione macro completa (read-only, env ignored) 
- exchanges/macro: cot.py + cot_contracts.py + fetchers.py copiati 1:1 con
  rewrite import mcp_common -> cerbero_mcp.common, mcp_macro -> cerbero_mcp.exchanges.macro
- nuovo MacroClient stateless wrapper: trasporta solo fred_api_key/finnhub_api_key,
  niente HTTP session (i fetchers usano async_client ad-hoc)
- tools.py: 11 tool (get_treasury_yields, get_yield_curve_slope,
  get_breakeven_inflation, get_economic_indicators, get_macro_calendar,
  get_market_overview, get_equity_futures, get_asset_price, get_cot_tff,
  get_cot_disaggregated, get_cot_extreme_positioning) — niente write,
  niente leverage_cap
- routers/macro.py: prefix /mcp-macro, 11 route POST /tools/*
- builder branch macro: stesse credenziali per testnet/mainnet (env ignorato);
  registry istanzia 2 entry, costo trascurabile (wrapper stateless)
- test migrati: test_cot.py + test_fetchers.py (test_server_acl.py skippato V1-only)
- nuovo test test_build_client_macro_no_env_distinction in test_exchanges_builder.py

Suite: 224 passed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-30 18:42:55 +02:00

403 lines
14 KiB
Python
Raw Blame History

from __future__ import annotations
from datetime import UTC
import httpx
import pytest
import pytest_httpx
from cerbero_mcp.exchanges.macro.fetchers import (
fetch_breakeven_inflation,
fetch_economic_indicators,
fetch_macro_calendar,
fetch_market_overview,
yield_curve_metrics,
)
# --- fetch_economic_indicators ---
@pytest.mark.asyncio
async def test_economic_indicators_no_key():
result = await fetch_economic_indicators(fred_api_key="")
assert "error" in result
assert result["error"] == "No FRED API key configured"
@pytest.mark.asyncio
async def test_economic_indicators_happy_path(httpx_mock: pytest_httpx.HTTPXMock):
for series_id in ("FEDFUNDS", "CPIAUCSL", "UNRATE", "DGS10"):
httpx_mock.add_response(
url=httpx.URL(
"https://api.stlouisfed.org/fred/series/observations",
params={
"series_id": series_id,
"api_key": "testkey",
"file_type": "json",
"sort_order": "desc",
"limit": "1",
},
),
json={"observations": [{"value": "5.25"}]},
)
result = await fetch_economic_indicators(fred_api_key="testkey")
assert result["fed_rate"] == 5.25
assert result["cpi"] == 5.25
assert result["unemployment"] == 5.25
assert result["us10y_yield"] == 5.25
assert "updated_at" in result
@pytest.mark.asyncio
async def test_economic_indicators_filter(httpx_mock: pytest_httpx.HTTPXMock):
httpx_mock.add_response(
url=httpx.URL(
"https://api.stlouisfed.org/fred/series/observations",
params={
"series_id": "FEDFUNDS",
"api_key": "k",
"file_type": "json",
"sort_order": "desc",
"limit": "1",
},
),
json={"observations": [{"value": "5.33"}]},
)
result = await fetch_economic_indicators(fred_api_key="k", indicators=["fed_rate"])
assert "fed_rate" in result
assert "cpi" not in result
# --- fetch_macro_calendar ---
@pytest.mark.asyncio
async def test_macro_calendar_forex_factory_happy(httpx_mock: pytest_httpx.HTTPXMock):
from datetime import datetime, timedelta
future = (datetime.now(UTC) + timedelta(days=1)).isoformat()
httpx_mock.add_response(
url="https://nfs.faireconomy.media/ff_calendar_thisweek.json",
json=[
{
"date": future,
"title": "CPI",
"country": "US",
"impact": "High",
"forecast": "3.0%",
"previous": "3.2%",
}
],
)
result = await fetch_macro_calendar()
assert "events" in result
assert len(result["events"]) >= 1
assert result["events"][0]["name"] == "CPI"
@pytest.mark.asyncio
async def test_macro_calendar_no_source(httpx_mock: pytest_httpx.HTTPXMock):
httpx_mock.add_response(
url="https://nfs.faireconomy.media/ff_calendar_thisweek.json",
status_code=500,
)
result = await fetch_macro_calendar(finnhub_api_key="")
assert result == {"events": [], "note": "No calendar source available"}
@pytest.mark.asyncio
@pytest.mark.httpx_mock(assert_all_responses_were_requested=False, assert_all_requests_were_expected=False)
async def test_macro_calendar_finnhub_fallback(httpx_mock: pytest_httpx.HTTPXMock):
httpx_mock.add_response(
url="https://nfs.faireconomy.media/ff_calendar_thisweek.json",
status_code=500,
)
def dispatch(request: httpx.Request) -> httpx.Response:
if "finnhub.io" in str(request.url):
return httpx.Response(
200,
json=[{"date": "2024-01-15", "event": "FOMC", "importance": "high", "forecast": "", "prev": ""}],
)
return httpx.Response(500)
httpx_mock.add_callback(dispatch)
result = await fetch_macro_calendar(finnhub_api_key="fkey")
assert "events" in result
assert result["events"][0]["name"] == "FOMC"
# --- fetch_market_overview ---
@pytest.mark.asyncio
async def test_market_overview_happy(httpx_mock: pytest_httpx.HTTPXMock):
httpx_mock.add_response(
url="https://api.coingecko.com/api/v3/global",
json={
"data": {
"market_cap_percentage": {"btc": 52.3},
"total_market_cap": {"usd": 2_000_000_000_000},
}
},
)
httpx_mock.add_response(
url=httpx.URL(
"https://api.coingecko.com/api/v3/simple/price",
params={"ids": "bitcoin,ethereum", "vs_currencies": "usd"},
),
json={"bitcoin": {"usd": 65000}, "ethereum": {"usd": 3500}},
)
import re as _re
httpx_mock.add_response(
url=_re.compile(
r"https://www\.deribit\.com/api/v2/public/get_volatility_index_data\?currency=BTC.*"
),
json={"result": {"data": [[1, 50, 52, 49, 51.5]], "continuation": None}},
)
httpx_mock.add_response(
url=_re.compile(
r"https://www\.deribit\.com/api/v2/public/get_volatility_index_data\?currency=ETH.*"
),
json={"result": {"data": [[1, 60, 62, 59, 61.2]], "continuation": None}},
)
import re as _re
httpx_mock.add_response(
url=_re.compile(r"https://query1\.finance\.yahoo\.com/v8/finance/chart/\^GSPC.*"),
json={"chart": {"result": [{"meta": {"regularMarketPrice": 5830.12}}]}},
)
httpx_mock.add_response(
url=_re.compile(r"https://query1\.finance\.yahoo\.com/v8/finance/chart/GC[%=].*"),
json={"chart": {"result": [{"meta": {"regularMarketPrice": 2412.5}}]}},
)
httpx_mock.add_response(
url=_re.compile(r"https://query1\.finance\.yahoo\.com/v8/finance/chart/\^VIX.*"),
json={"chart": {"result": [{"meta": {"regularMarketPrice": 18.3}}]}},
)
# Clear module cache to force fresh fetch
from cerbero_mcp.exchanges.macro import fetchers as _f
_f._MARKET_CACHE["data"] = None
_f._MARKET_CACHE["ts"] = 0.0
result = await fetch_market_overview()
assert result["btc_dominance"] == 52.3
assert result["btc_price"] == 65000
assert result["eth_price"] == 3500
assert result["total_market_cap"] == 2_000_000_000_000
assert result["dvol_btc"] == 51.5
assert result["dvol_eth"] == 61.2
assert result["sp500"] == 5830.12
assert result["gold"] == 2412.5
assert result["vix"] == 18.3
assert "data_timestamp" in result
# --- yield_curve_metrics ---
def test_yield_curve_metrics_normal_curve():
out = yield_curve_metrics({"us2y": 4.0, "us5y": 4.2, "us10y": 4.5, "us30y": 4.8})
assert out["slope_2y10y"] == 0.5
assert out["slope_5y30y"] == 0.6
assert out["regime"] == "steep"
# butterfly: 2*4.5 - 4.0 - 4.8 = 0.2
assert out["butterfly_2_10_30"] == 0.2
def test_yield_curve_metrics_inverted():
out = yield_curve_metrics({"us2y": 5.5, "us5y": 5.0, "us10y": 4.5, "us30y": 4.3})
assert out["slope_2y10y"] == -1.0
assert out["regime"] == "inverted"
def test_yield_curve_metrics_partial_data():
out = yield_curve_metrics({"us10y": 4.5})
assert out["slope_2y10y"] is None
assert out["regime"] == "unknown"
# --- fetch_breakeven_inflation ---
@pytest.mark.asyncio
async def test_breakeven_no_key():
out = await fetch_breakeven_inflation(fred_api_key="")
assert "error" in out
@pytest.mark.asyncio
async def test_breakeven_happy_path(httpx_mock: pytest_httpx.HTTPXMock):
for series_id, val in [("T5YIE", "2.3"), ("T10YIE", "2.5"), ("T5YIFR", "2.7")]:
httpx_mock.add_response(
url=httpx.URL(
"https://api.stlouisfed.org/fred/series/observations",
params={
"series_id": series_id,
"api_key": "k",
"file_type": "json",
"sort_order": "desc",
"limit": "1",
},
),
json={"observations": [{"value": val}]},
)
out = await fetch_breakeven_inflation(fred_api_key="k")
assert out["breakevens"]["be_5y"] == 2.3
assert out["breakevens"]["be_10y"] == 2.5
assert out["breakevens"]["be_5y5y_forward"] == 2.7
assert out["interpretation"] == "anchored"
@pytest.mark.asyncio
async def test_breakeven_high_inflation(httpx_mock: pytest_httpx.HTTPXMock):
for series_id in ("T5YIE", "T10YIE", "T5YIFR"):
httpx_mock.add_response(
url=httpx.URL(
"https://api.stlouisfed.org/fred/series/observations",
params={
"series_id": series_id,
"api_key": "k",
"file_type": "json",
"sort_order": "desc",
"limit": "1",
},
),
json={"observations": [{"value": "3.5"}]},
)
out = await fetch_breakeven_inflation(fred_api_key="k")
assert out["interpretation"] == "high_inflation_expected"
@pytest.mark.asyncio
async def test_fetch_cot_tff_happy_path(httpx_mock: pytest_httpx.HTTPXMock):
from cerbero_mcp.exchanges.macro.fetchers import fetch_cot_tff
httpx_mock.add_response(
url=httpx.URL(
"https://publicreporting.cftc.gov/resource/gpe5-46if.json",
params={
"cftc_contract_market_code": "13874A",
"$order": "report_date_as_yyyy_mm_dd DESC",
"$limit": "52",
},
),
json=[
{
"report_date_as_yyyy_mm_dd": "2026-04-22T00:00:00.000",
"dealer_positions_long_all": "12345",
"dealer_positions_short_all": "23456",
"asset_mgr_positions_long": "654321",
"asset_mgr_positions_short": "200000",
"lev_money_positions_long": "100000",
"lev_money_positions_short": "350000",
"other_rept_positions_long": "50000",
"other_rept_positions_short": "50000",
"open_interest_all": "2500000",
},
{
"report_date_as_yyyy_mm_dd": "2026-04-15T00:00:00.000",
"dealer_positions_long_all": "11000",
"dealer_positions_short_all": "22000",
"asset_mgr_positions_long": "640000",
"asset_mgr_positions_short": "210000",
"lev_money_positions_long": "110000",
"lev_money_positions_short": "320000",
"other_rept_positions_long": "48000",
"other_rept_positions_short": "52000",
"open_interest_all": "2480000",
},
],
)
out = await fetch_cot_tff("ES", lookback_weeks=52)
assert out["symbol"] == "ES"
assert out["report_type"] == "tff"
assert len(out["rows"]) == 2
# Ordering ASC by date (oldest first)
assert out["rows"][0]["report_date"] == "2026-04-15"
assert out["rows"][1]["report_date"] == "2026-04-22"
assert out["rows"][1]["lev_funds_net"] == -250000
assert "data_timestamp" in out
@pytest.mark.asyncio
async def test_fetch_cot_tff_unknown_symbol():
from cerbero_mcp.exchanges.macro.fetchers import fetch_cot_tff
out = await fetch_cot_tff("INVALID", lookback_weeks=52)
assert out.get("error") == "unknown_symbol"
assert "ES" in out.get("available", [])
@pytest.mark.asyncio
async def test_fetch_cot_disagg_happy_path(httpx_mock: pytest_httpx.HTTPXMock):
from cerbero_mcp.exchanges.macro.fetchers import fetch_cot_disaggregated
httpx_mock.add_response(
url=httpx.URL(
"https://publicreporting.cftc.gov/resource/72hh-3qpy.json",
params={
"cftc_contract_market_code": "067651",
"$order": "report_date_as_yyyy_mm_dd DESC",
"$limit": "52",
},
),
json=[
{
"report_date_as_yyyy_mm_dd": "2026-04-22T00:00:00.000",
"prod_merc_positions_long_all": "100000",
"prod_merc_positions_short_all": "300000",
"swap_positions_long_all": "50000",
"swap_positions_short_all": "60000",
"m_money_positions_long_all": "200000",
"m_money_positions_short_all": "80000",
"other_rept_positions_long_all": "10000",
"other_rept_positions_short_all": "10000",
"open_interest_all": "1500000",
},
],
)
out = await fetch_cot_disaggregated("CL", lookback_weeks=52)
assert out["symbol"] == "CL"
assert out["report_type"] == "disaggregated"
assert len(out["rows"]) == 1
assert out["rows"][0]["managed_money_net"] == 120000
assert out["rows"][0]["producer_net"] == -200000
@pytest.mark.asyncio
async def test_fetch_cot_disagg_unknown_symbol():
from cerbero_mcp.exchanges.macro.fetchers import fetch_cot_disaggregated
out = await fetch_cot_disaggregated("XYZ", lookback_weeks=52)
assert out.get("error") == "unknown_symbol"
assert "CL" in out.get("available", [])
@pytest.mark.asyncio
async def test_fetch_cot_extreme_positioning_flags_outliers(monkeypatch):
"""Mock fetch_cot_tff e fetch_cot_disagg per simulare history e ultimo punto."""
from unittest.mock import AsyncMock
from cerbero_mcp.exchanges.macro import fetchers as f
# Simula una serie ES dove ultimo lev_funds_net è in basso (extreme_short)
es_rows = [
{"report_date": f"2026-{m:02d}-01", "lev_funds_net": v}
for m, v in [(1, 0), (2, 50), (3, 100), (4, -500)]
]
cl_rows = [
{"report_date": f"2026-{m:02d}-01", "managed_money_net": v}
for m, v in [(1, 100), (2, 200), (3, 300), (4, 1000)]
]
async def fake_tff(symbol, lookback_weeks):
if symbol == "ES":
return {"symbol": "ES", "report_type": "tff", "rows": es_rows}
return {"symbol": symbol, "report_type": "tff", "rows": []}
async def fake_disagg(symbol, lookback_weeks):
if symbol == "CL":
return {"symbol": "CL", "report_type": "disaggregated", "rows": cl_rows}
return {"symbol": symbol, "report_type": "disaggregated", "rows": []}
monkeypatch.setattr(f, "fetch_cot_tff", AsyncMock(side_effect=fake_tff))
monkeypatch.setattr(f, "fetch_cot_disaggregated", AsyncMock(side_effect=fake_disagg))
out = await f.fetch_cot_extreme_positioning(lookback_weeks=4)
assert "extremes" in out
by_sym = {e["symbol"]: e for e in out["extremes"]}
assert by_sym["ES"]["signal"] == "extreme_short"
assert by_sym["ES"]["key_role"] == "lev_funds"
assert by_sym["CL"]["signal"] == "extreme_long"
assert by_sym["CL"]["key_role"] == "managed_money"
Reference in New Issue View Git Blame Copy Permalink