131 lines
4.3 KiB
Python
131 lines
4.3 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 (
|
|
GateTargetLadderResponse,
|
|
GateTargetLevelResult,
|
|
SRLevelResponse,
|
|
SRLevelResult,
|
|
SRZoneResult,
|
|
)
|
|
from app.services.price_service import query_ohlcv
|
|
from app.services.sr_service import (
|
|
cluster_sr_zones,
|
|
detect_gate_target_ladder,
|
|
get_sr_levels,
|
|
)
|
|
|
|
router = APIRouter(tags=["sr-levels"])
|
|
|
|
|
|
@router.get("/gate-target-ladder/{symbol}", response_model=APIEnvelope)
|
|
async def read_gate_target_ladder(
|
|
symbol: str,
|
|
_user=Depends(require_access),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> APIEnvelope:
|
|
"""Return the transient, volume-free GTL for chart diagnostics.
|
|
|
|
These proposals are not persisted ``SRLevel`` rows and must not be
|
|
presented as structural support/resistance.
|
|
"""
|
|
records = await query_ohlcv(db, symbol)
|
|
if not records:
|
|
data = GateTargetLadderResponse(
|
|
symbol=symbol.upper(),
|
|
levels=[],
|
|
count=0,
|
|
lookback_bars=0,
|
|
)
|
|
return APIEnvelope(status="success", data=data.model_dump())
|
|
|
|
highs = [float(record.high) for record in records]
|
|
lows = [float(record.low) for record in records]
|
|
closes = [float(record.close) for record in records]
|
|
detected = detect_gate_target_ladder(highs, lows, closes)
|
|
levels = [
|
|
GateTargetLevelResult(
|
|
price_level=float(level["price_level"]),
|
|
type=level["type"],
|
|
strength=int(level["strength"]),
|
|
detection_method=str(level.get("detection_method", "unknown")),
|
|
sources=list(level.get("sources") or []),
|
|
traffic_count=int(level.get("rejection_count", 0) or 0),
|
|
)
|
|
for level in sorted(detected, key=lambda row: float(row["price_level"]))
|
|
]
|
|
data = GateTargetLadderResponse(
|
|
symbol=symbol.upper(),
|
|
levels=levels,
|
|
count=len(levels),
|
|
lookback_bars=len(records),
|
|
)
|
|
return APIEnvelope(status="success", data=data.model_dump())
|
|
|
|
|
|
@router.get("/sr-levels/{symbol}", response_model=APIEnvelope)
|
|
async def read_sr_levels(
|
|
symbol: str,
|
|
tolerance: float | None = Query(
|
|
None,
|
|
ge=0,
|
|
le=0.1,
|
|
description="Merge tolerance as fraction of price; omit for ATR-adaptive default",
|
|
),
|
|
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())
|