feat: add portfolio optimization and congress tracking (TDD)
Portfolio optimization (3 endpoints): - POST /portfolio/optimize - HRP optimal weights via scipy clustering - POST /portfolio/correlation - pairwise correlation matrix - POST /portfolio/risk-parity - inverse-volatility risk parity weights Congress tracking (2 endpoints): - GET /regulators/congress/trades - congress member stock trades - GET /regulators/congress/bills?query= - search congress bills Implementation: - portfolio_service.py: HRP with scipy fallback to inverse-vol - congress_service.py: multi-provider fallback pattern - 51 new tests (14 portfolio unit, 20 portfolio route, 12 congress unit, 7 congress route) - All 312 tests passing
This commit is contained in:
187
tests/test_congress_service.py
Normal file
187
tests/test_congress_service.py
Normal file
@@ -0,0 +1,187 @@
|
||||
"""Tests for congress trading service (TDD - RED phase first)."""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# --- get_congress_trades ---
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_congress_trades_happy_path():
|
||||
"""Returns list of trade dicts when OBB call succeeds."""
|
||||
expected = [
|
||||
{
|
||||
"representative": "Nancy Pelosi",
|
||||
"ticker": "NVDA",
|
||||
"transaction_date": "2024-01-15",
|
||||
"transaction_type": "Purchase",
|
||||
"amount": "$1,000,001-$5,000,000",
|
||||
}
|
||||
]
|
||||
|
||||
import congress_service
|
||||
|
||||
mock_fn = MagicMock()
|
||||
with patch.object(congress_service, "_get_congress_fn", return_value=mock_fn), \
|
||||
patch.object(congress_service, "_try_obb_call", new_callable=AsyncMock, return_value=expected):
|
||||
result = await congress_service.get_congress_trades()
|
||||
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
assert result[0]["representative"] == "Nancy Pelosi"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_congress_trades_returns_empty_when_fn_not_available():
|
||||
"""Returns empty list when OBB congress function is not available."""
|
||||
import congress_service
|
||||
|
||||
with patch.object(congress_service, "_get_congress_fn", return_value=None):
|
||||
result = await congress_service.get_congress_trades()
|
||||
|
||||
assert result == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_congress_trades_returns_empty_on_all_provider_failures():
|
||||
"""Returns empty list when all providers fail (_try_obb_call returns None)."""
|
||||
import congress_service
|
||||
|
||||
mock_fn = MagicMock()
|
||||
with patch.object(congress_service, "_get_congress_fn", return_value=mock_fn), \
|
||||
patch.object(congress_service, "_try_obb_call", new_callable=AsyncMock, return_value=None):
|
||||
result = await congress_service.get_congress_trades()
|
||||
|
||||
assert result == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_congress_trades_empty_list_result():
|
||||
"""Returns empty list when _try_obb_call returns empty list."""
|
||||
import congress_service
|
||||
|
||||
mock_fn = MagicMock()
|
||||
with patch.object(congress_service, "_get_congress_fn", return_value=mock_fn), \
|
||||
patch.object(congress_service, "_try_obb_call", new_callable=AsyncMock, return_value=[]):
|
||||
result = await congress_service.get_congress_trades()
|
||||
|
||||
assert result == []
|
||||
|
||||
|
||||
# --- _get_congress_fn ---
|
||||
|
||||
|
||||
def test_get_congress_fn_returns_none_when_attribute_missing():
|
||||
"""Returns None gracefully when obb.regulators.government_us is absent."""
|
||||
import congress_service
|
||||
|
||||
mock_obb = MagicMock(spec=[]) # spec with no attributes
|
||||
with patch.object(congress_service, "obb", mock_obb):
|
||||
result = congress_service._get_congress_fn()
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_get_congress_fn_returns_callable_when_available():
|
||||
"""Returns the congress_trading callable when attribute exists."""
|
||||
import congress_service
|
||||
|
||||
mock_fn = MagicMock()
|
||||
mock_obb = MagicMock()
|
||||
mock_obb.regulators.government_us.congress_trading = mock_fn
|
||||
|
||||
with patch.object(congress_service, "obb", mock_obb):
|
||||
result = congress_service._get_congress_fn()
|
||||
|
||||
assert result is mock_fn
|
||||
|
||||
|
||||
# --- _try_obb_call ---
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_try_obb_call_returns_list_on_success():
|
||||
"""_try_obb_call converts OBBject result to list via to_list."""
|
||||
import congress_service
|
||||
|
||||
mock_result = MagicMock()
|
||||
expected = [{"ticker": "AAPL"}]
|
||||
|
||||
with patch.object(congress_service, "to_list", return_value=expected), \
|
||||
patch("congress_service.asyncio.to_thread", new_callable=AsyncMock, return_value=mock_result):
|
||||
result = await congress_service._try_obb_call(MagicMock())
|
||||
|
||||
assert result == expected
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_try_obb_call_returns_none_on_exception():
|
||||
"""_try_obb_call returns None when asyncio.to_thread raises."""
|
||||
import congress_service
|
||||
|
||||
with patch("congress_service.asyncio.to_thread", new_callable=AsyncMock, side_effect=Exception("fail")):
|
||||
result = await congress_service._try_obb_call(MagicMock())
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
# --- search_congress_bills ---
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_congress_bills_happy_path():
|
||||
"""Returns list of bill dicts when OBB call succeeds."""
|
||||
expected = [
|
||||
{"title": "Infrastructure Investment and Jobs Act", "bill_id": "HR3684"},
|
||||
{"title": "Inflation Reduction Act", "bill_id": "HR5376"},
|
||||
]
|
||||
|
||||
import congress_service
|
||||
|
||||
mock_fn = MagicMock()
|
||||
with patch.object(congress_service, "_get_congress_fn", return_value=mock_fn), \
|
||||
patch.object(congress_service, "_try_obb_call", new_callable=AsyncMock, return_value=expected):
|
||||
result = await congress_service.search_congress_bills("infrastructure")
|
||||
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 2
|
||||
assert result[0]["bill_id"] == "HR3684"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_congress_bills_returns_empty_when_fn_not_available():
|
||||
"""Returns empty list when OBB function is not available."""
|
||||
import congress_service
|
||||
|
||||
with patch.object(congress_service, "_get_congress_fn", return_value=None):
|
||||
result = await congress_service.search_congress_bills("taxes")
|
||||
|
||||
assert result == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_congress_bills_returns_empty_on_failure():
|
||||
"""Returns empty list when all providers fail."""
|
||||
import congress_service
|
||||
|
||||
mock_fn = MagicMock()
|
||||
with patch.object(congress_service, "_get_congress_fn", return_value=mock_fn), \
|
||||
patch.object(congress_service, "_try_obb_call", new_callable=AsyncMock, return_value=None):
|
||||
result = await congress_service.search_congress_bills("taxes")
|
||||
|
||||
assert result == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_congress_bills_empty_results():
|
||||
"""Returns empty list when _try_obb_call returns empty list."""
|
||||
import congress_service
|
||||
|
||||
mock_fn = MagicMock()
|
||||
with patch.object(congress_service, "_get_congress_fn", return_value=mock_fn), \
|
||||
patch.object(congress_service, "_try_obb_call", new_callable=AsyncMock, return_value=[]):
|
||||
result = await congress_service.search_congress_bills("nonexistent")
|
||||
|
||||
assert result == []
|
||||
Reference in New Issue
Block a user