Add GTL price-traffic chart diagnostic
This commit is contained in:
@@ -5,13 +5,68 @@ 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.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, get_sr_levels
|
||||
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,
|
||||
|
||||
@@ -40,3 +40,23 @@ class SRLevelResponse(BaseModel):
|
||||
zones: list[SRZoneResult] = []
|
||||
visible_levels: list[SRLevelResult] = []
|
||||
count: int
|
||||
|
||||
|
||||
class GateTargetLevelResult(BaseModel):
|
||||
"""A transient Gate Target Ladder proposal for diagnostic display."""
|
||||
|
||||
price_level: float
|
||||
type: Literal["support", "resistance"]
|
||||
strength: int = Field(ge=0, le=100)
|
||||
detection_method: str
|
||||
sources: list[str] = Field(default_factory=list)
|
||||
traffic_count: int = Field(ge=0)
|
||||
|
||||
|
||||
class GateTargetLadderResponse(BaseModel):
|
||||
"""Volume-free Gate Target Ladder computed from current OHLCV history."""
|
||||
|
||||
symbol: str
|
||||
levels: list[GateTargetLevelResult]
|
||||
count: int
|
||||
lookback_bars: int
|
||||
|
||||
Reference in New Issue
Block a user