71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
"""S/R Levels router — support/resistance detection endpoints."""
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.dependencies import get_db, require_access
|
|
from app.schemas.common import APIEnvelope
|
|
from app.schemas.sr_level import SRLevelResponse, SRLevelResult, SRZoneResult
|
|
from app.services.price_service import query_ohlcv
|
|
from app.services.sr_service import cluster_sr_zones, get_sr_levels
|
|
|
|
router = APIRouter(tags=["sr-levels"])
|
|
|
|
|
|
@router.get("/sr-levels/{symbol}", response_model=APIEnvelope)
|
|
async def read_sr_levels(
|
|
symbol: str,
|
|
tolerance: float = Query(0.005, ge=0, le=0.1, description="Merge tolerance (default 0.5%)"),
|
|
max_zones: int = Query(6, ge=0, description="Max S/R zones to return (default 6)"),
|
|
_user=Depends(require_access),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> APIEnvelope:
|
|
"""Get support/resistance levels for a symbol, sorted by strength descending."""
|
|
levels = await get_sr_levels(db, symbol, tolerance)
|
|
|
|
level_results = [
|
|
SRLevelResult(
|
|
id=lvl.id,
|
|
price_level=lvl.price_level,
|
|
type=lvl.type,
|
|
strength=lvl.strength,
|
|
detection_method=lvl.detection_method,
|
|
created_at=lvl.created_at,
|
|
)
|
|
for lvl in levels
|
|
]
|
|
|
|
# Compute S/R zones from the fetched levels
|
|
zones: list[SRZoneResult] = []
|
|
if levels and max_zones > 0:
|
|
# Get current price from latest OHLCV close
|
|
ohlcv_records = await query_ohlcv(db, symbol)
|
|
if ohlcv_records:
|
|
current_price = ohlcv_records[-1].close
|
|
level_dicts = [
|
|
{"price_level": lvl.price_level, "strength": lvl.strength}
|
|
for lvl in levels
|
|
]
|
|
raw_zones = cluster_sr_zones(
|
|
level_dicts, current_price, tolerance=0.02, max_zones=max_zones
|
|
)
|
|
zones = [SRZoneResult(**z) for z in raw_zones]
|
|
|
|
# Filter levels to only those within at least one zone's [low, high] range
|
|
visible_levels: list[SRLevelResult] = []
|
|
if zones:
|
|
visible_levels = [
|
|
lvl
|
|
for lvl in level_results
|
|
if any(z.low <= lvl.price_level <= z.high for z in zones)
|
|
]
|
|
|
|
data = SRLevelResponse(
|
|
symbol=symbol.upper(),
|
|
levels=level_results,
|
|
zones=zones,
|
|
visible_levels=visible_levels,
|
|
count=len(levels),
|
|
)
|
|
return APIEnvelope(status="success", data=data.model_dump())
|