98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Annotated, Literal
|
|
|
|
from fastapi import APIRouter, Path, Query, Request
|
|
|
|
from backend.features.accounts.auth import AdminWritePrincipal, AuthenticatedPrincipal
|
|
from backend.features.market.schemas import (
|
|
ChartResponse,
|
|
MarketSummaryResponse,
|
|
MarketWorkspaceResponse,
|
|
ReferenceSyncResponse,
|
|
RotationMembersResponse,
|
|
SearchResponse,
|
|
SnapshotSyncResponse,
|
|
TradeContextResponse,
|
|
)
|
|
|
|
router = APIRouter(prefix="/market", tags=["market"])
|
|
|
|
|
|
@router.get("/context", response_model=TradeContextResponse)
|
|
def context(
|
|
request: Request,
|
|
_principal: AuthenticatedPrincipal,
|
|
requested_date: Annotated[str | None, Query(alias="date")] = None,
|
|
) -> dict:
|
|
return request.app.state.container.market.context(requested_date)
|
|
|
|
|
|
@router.get("/summary", response_model=MarketSummaryResponse)
|
|
def summary(
|
|
request: Request,
|
|
_principal: AuthenticatedPrincipal,
|
|
requested_date: Annotated[str | None, Query(alias="date")] = None,
|
|
) -> dict:
|
|
return request.app.state.container.market.summary(requested_date)
|
|
|
|
|
|
@router.get("/search", response_model=SearchResponse)
|
|
def search(
|
|
request: Request,
|
|
_principal: AuthenticatedPrincipal,
|
|
query: Annotated[str, Query(alias="q", max_length=80)] = "",
|
|
) -> dict:
|
|
return request.app.state.container.market.search(query)
|
|
|
|
|
|
@router.get("/entities/{entity_type}/{identifier}/charts/{interval}", response_model=ChartResponse)
|
|
def chart(
|
|
request: Request,
|
|
_principal: AuthenticatedPrincipal,
|
|
entity_type: Annotated[Literal["stock", "sector", "theme", "index"], Path()],
|
|
identifier: Annotated[str, Path(min_length=1, max_length=40)],
|
|
interval: Annotated[Literal["day", "minute"], Path()],
|
|
) -> dict:
|
|
return request.app.state.container.market.chart(entity_type, identifier, interval)
|
|
|
|
|
|
@router.post("/reference-sync", response_model=ReferenceSyncResponse)
|
|
def refresh_reference(request: Request, _principal: AdminWritePrincipal) -> dict[str, int | str]:
|
|
return request.app.state.container.market.refresh_reference()
|
|
|
|
|
|
@router.post("/snapshot-sync", response_model=SnapshotSyncResponse)
|
|
def sync_snapshot(
|
|
request: Request,
|
|
_principal: AdminWritePrincipal,
|
|
requested_date: Annotated[str | None, Query(alias="date")] = None,
|
|
) -> dict:
|
|
return request.app.state.container.market.sync_snapshot(requested_date)
|
|
|
|
|
|
@router.get("/workspaces/{key}", response_model=MarketWorkspaceResponse)
|
|
def workspace(
|
|
request: Request,
|
|
_principal: AuthenticatedPrincipal,
|
|
key: Annotated[
|
|
Literal[
|
|
"emotion", "pool", "broken", "limit-down", "yesterday", "performance",
|
|
"ladder", "rotation",
|
|
],
|
|
Path(),
|
|
],
|
|
requested_date: Annotated[str | None, Query(alias="date")] = None,
|
|
) -> dict:
|
|
return request.app.state.container.market.workspace(key, requested_date)
|
|
|
|
|
|
@router.get("/rotation-members", response_model=RotationMembersResponse)
|
|
def rotation_members(
|
|
request: Request,
|
|
_principal: AuthenticatedPrincipal,
|
|
sector_name: Annotated[str, Query(alias="sector", min_length=1, max_length=60)],
|
|
requested_date: Annotated[str | None, Query(alias="date")] = None,
|
|
) -> dict:
|
|
return request.app.state.container.market.rotation_members(sector_name, requested_date)
|