310 lines
11 KiB
Python
310 lines
11 KiB
Python
"""Unit tests for the S/R levels router — zone integration."""
|
|
|
|
from datetime import datetime
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.middleware import register_exception_handlers
|
|
from app.routers.sr_levels import router
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FakeLevel:
|
|
"""Mimics an SRLevel ORM model."""
|
|
|
|
def __init__(self, id, price_level, type, strength, detection_method):
|
|
self.id = id
|
|
self.price_level = price_level
|
|
self.type = type
|
|
self.strength = strength
|
|
self.detection_method = detection_method
|
|
self.created_at = datetime(2024, 1, 1)
|
|
|
|
|
|
class _FakeOHLCV:
|
|
"""Mimics an OHLCVRecord with price attributes."""
|
|
|
|
def __init__(self, close: float, high: float | None = None, low: float | None = None):
|
|
self.close = close
|
|
self.high = high if high is not None else close + 1.0
|
|
self.low = low if low is not None else close - 1.0
|
|
|
|
|
|
def _make_app() -> FastAPI:
|
|
app = FastAPI()
|
|
register_exception_handlers(app)
|
|
app.include_router(router, prefix="/api/v1")
|
|
|
|
# Override auth dependency to no-op
|
|
from app.dependencies import require_access, get_db
|
|
|
|
app.dependency_overrides[require_access] = lambda: None
|
|
app.dependency_overrides[get_db] = lambda: AsyncMock()
|
|
return app
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
SAMPLE_LEVELS = [
|
|
_FakeLevel(1, 95.0, "support", 60, "volume_profile"),
|
|
_FakeLevel(2, 96.0, "support", 40, "pivot_point"),
|
|
_FakeLevel(3, 110.0, "resistance", 80, "merged"),
|
|
]
|
|
|
|
SAMPLE_OHLCV = [_FakeOHLCV(100.0)]
|
|
|
|
|
|
class TestGateTargetLadderRouter:
|
|
@patch("app.routers.sr_levels.detect_gate_target_ladder")
|
|
@patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock)
|
|
def test_returns_transient_price_traffic_proposals(self, mock_ohlcv, mock_detect):
|
|
mock_ohlcv.return_value = [
|
|
_FakeOHLCV(100.0, high=101.0, low=99.0),
|
|
_FakeOHLCV(102.0, high=103.0, low=100.0),
|
|
]
|
|
mock_detect.return_value = [
|
|
{
|
|
"price_level": 105.0,
|
|
"type": "resistance",
|
|
"strength": 80,
|
|
"detection_method": "merged",
|
|
"sources": ["pivot_point", "range_grid"],
|
|
"rejection_count": 7,
|
|
},
|
|
{
|
|
"price_level": 95.0,
|
|
"type": "support",
|
|
"strength": 60,
|
|
"detection_method": "range_grid",
|
|
"sources": ["range_grid"],
|
|
"rejection_count": 4,
|
|
},
|
|
]
|
|
|
|
response = TestClient(_make_app()).get("/api/v1/gate-target-ladder/aapl")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()["data"]
|
|
assert data["symbol"] == "AAPL"
|
|
assert data["lookback_bars"] == 2
|
|
assert [level["price_level"] for level in data["levels"]] == [95.0, 105.0]
|
|
assert data["levels"][1] == {
|
|
"price_level": 105.0,
|
|
"type": "resistance",
|
|
"strength": 80,
|
|
"detection_method": "merged",
|
|
"sources": ["pivot_point", "range_grid"],
|
|
"traffic_count": 7,
|
|
}
|
|
mock_detect.assert_called_once_with(
|
|
[101.0, 103.0],
|
|
[99.0, 100.0],
|
|
[100.0, 102.0],
|
|
)
|
|
|
|
@patch("app.routers.sr_levels.detect_gate_target_ladder")
|
|
@patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock)
|
|
def test_empty_history_returns_empty_ladder(self, mock_ohlcv, mock_detect):
|
|
mock_ohlcv.return_value = []
|
|
|
|
response = TestClient(_make_app()).get("/api/v1/gate-target-ladder/AAPL")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["data"] == {
|
|
"symbol": "AAPL",
|
|
"levels": [],
|
|
"count": 0,
|
|
"lookback_bars": 0,
|
|
}
|
|
mock_detect.assert_not_called()
|
|
|
|
|
|
class TestSRLevelsRouterZones:
|
|
"""Tests for max_zones parameter and zone inclusion in response."""
|
|
|
|
@patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock)
|
|
@patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock)
|
|
def test_default_max_zones_returns_zones(self, mock_get_sr, mock_ohlcv):
|
|
mock_get_sr.return_value = SAMPLE_LEVELS
|
|
mock_ohlcv.return_value = SAMPLE_OHLCV
|
|
|
|
app = _make_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/api/v1/sr-levels/AAPL")
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["status"] == "success"
|
|
data = body["data"]
|
|
assert "zones" in data
|
|
assert isinstance(data["zones"], list)
|
|
# With default max_zones=6, we should get zones
|
|
assert len(data["zones"]) > 0
|
|
|
|
@patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock)
|
|
@patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock)
|
|
def test_max_zones_zero_returns_empty_zones(self, mock_get_sr, mock_ohlcv):
|
|
mock_get_sr.return_value = SAMPLE_LEVELS
|
|
mock_ohlcv.return_value = SAMPLE_OHLCV
|
|
|
|
app = _make_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/api/v1/sr-levels/AAPL?max_zones=0")
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()["data"]
|
|
assert data["zones"] == []
|
|
|
|
@patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock)
|
|
@patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock)
|
|
def test_max_zones_limits_zone_count(self, mock_get_sr, mock_ohlcv):
|
|
mock_get_sr.return_value = SAMPLE_LEVELS
|
|
mock_ohlcv.return_value = SAMPLE_OHLCV
|
|
|
|
app = _make_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/api/v1/sr-levels/AAPL?max_zones=1")
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()["data"]
|
|
assert len(data["zones"]) <= 1
|
|
|
|
@patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock)
|
|
@patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock)
|
|
def test_no_ohlcv_data_returns_empty_zones(self, mock_get_sr, mock_ohlcv):
|
|
mock_get_sr.return_value = SAMPLE_LEVELS
|
|
mock_ohlcv.return_value = [] # No OHLCV data
|
|
|
|
app = _make_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/api/v1/sr-levels/AAPL")
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()["data"]
|
|
assert data["zones"] == []
|
|
# Levels should still be present
|
|
assert len(data["levels"]) == 3
|
|
|
|
@patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock)
|
|
@patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock)
|
|
def test_no_levels_returns_empty_zones(self, mock_get_sr, mock_ohlcv):
|
|
mock_get_sr.return_value = []
|
|
mock_ohlcv.return_value = SAMPLE_OHLCV
|
|
|
|
app = _make_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/api/v1/sr-levels/AAPL")
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()["data"]
|
|
assert data["zones"] == []
|
|
assert data["levels"] == []
|
|
assert data["count"] == 0
|
|
|
|
@patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock)
|
|
@patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock)
|
|
def test_zone_fields_present(self, mock_get_sr, mock_ohlcv):
|
|
mock_get_sr.return_value = SAMPLE_LEVELS
|
|
mock_ohlcv.return_value = SAMPLE_OHLCV
|
|
|
|
app = _make_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/api/v1/sr-levels/AAPL")
|
|
|
|
data = resp.json()["data"]
|
|
for zone in data["zones"]:
|
|
assert "low" in zone
|
|
assert "high" in zone
|
|
assert "midpoint" in zone
|
|
assert "strength" in zone
|
|
assert "type" in zone
|
|
assert "level_count" in zone
|
|
assert zone["type"] in ("support", "resistance")
|
|
|
|
|
|
class TestSRLevelsRouterVisibleLevels:
|
|
"""Tests for visible_levels filtering in the SR levels response."""
|
|
|
|
@patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock)
|
|
@patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock)
|
|
def test_visible_levels_present_in_response(self, mock_get_sr, mock_ohlcv):
|
|
"""visible_levels field is always present in the API response."""
|
|
mock_get_sr.return_value = SAMPLE_LEVELS
|
|
mock_ohlcv.return_value = SAMPLE_OHLCV
|
|
|
|
app = _make_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/api/v1/sr-levels/AAPL")
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()["data"]
|
|
assert "visible_levels" in data
|
|
assert isinstance(data["visible_levels"], list)
|
|
|
|
@patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock)
|
|
@patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock)
|
|
def test_visible_levels_within_zone_bounds(self, mock_get_sr, mock_ohlcv):
|
|
"""Every visible level has a price within at least one zone's [low, high] range."""
|
|
mock_get_sr.return_value = SAMPLE_LEVELS
|
|
mock_ohlcv.return_value = SAMPLE_OHLCV
|
|
|
|
app = _make_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/api/v1/sr-levels/AAPL")
|
|
|
|
data = resp.json()["data"]
|
|
zones = data["zones"]
|
|
visible = data["visible_levels"]
|
|
|
|
# When zones exist, each visible level must fall within a zone
|
|
for lvl in visible:
|
|
price = lvl["price_level"]
|
|
assert any(
|
|
z["low"] <= price <= z["high"] for z in zones
|
|
), f"visible level price {price} not within any zone bounds"
|
|
|
|
# visible_levels must be a subset of levels (by id)
|
|
level_ids = {level["id"] for level in data["levels"]}
|
|
for lvl in visible:
|
|
assert lvl["id"] in level_ids
|
|
|
|
@patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock)
|
|
@patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock)
|
|
def test_visible_levels_empty_when_no_ohlcv(self, mock_get_sr, mock_ohlcv):
|
|
"""visible_levels is empty when no OHLCV data exists (zones are empty)."""
|
|
mock_get_sr.return_value = SAMPLE_LEVELS
|
|
mock_ohlcv.return_value = []
|
|
|
|
app = _make_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/api/v1/sr-levels/AAPL")
|
|
|
|
data = resp.json()["data"]
|
|
assert data["zones"] == []
|
|
assert data["visible_levels"] == []
|
|
|
|
@patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock)
|
|
@patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock)
|
|
def test_visible_levels_empty_when_max_zones_zero(self, mock_get_sr, mock_ohlcv):
|
|
"""visible_levels is empty when max_zones=0 (zones are empty)."""
|
|
mock_get_sr.return_value = SAMPLE_LEVELS
|
|
mock_ohlcv.return_value = SAMPLE_OHLCV
|
|
|
|
app = _make_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/api/v1/sr-levels/AAPL?max_zones=0")
|
|
|
|
data = resp.json()["data"]
|
|
assert data["zones"] == []
|
|
assert data["visible_levels"] == []
|
|
|