routes/data.py

from services.indicator import Indicator
from models.ichimoku import IchimokuPayload
from models.rsi import RSIPayload
from models.price import PricePayload
from models.macd import MACDPayload
from models.index import IndexPayload
from fastapi import status, APIRouter
from typing import Sequence
import json

router = APIRouter()


@router.get(
    "/update_daily_price",
    name="Update daily price data",
    status_code=status.HTTP_200_OK
)
async def update_daily_price():
    Indicator.update_daily_price()


@router.get(
    "/update_entire_price",
    name="Update entire price data",
    status_code=status.HTTP_200_OK
)
async def update_entire_price():
    Indicator.update_entire_price()


@router.get(
    "/update_daily_rsi",
    name="Update daily rsi data",
    status_code=status.HTTP_200_OK
)
async def update_daily_rsi():
    Indicator.update_daily_rsi()


@router.get(
    "/update_entire_rsi",
    name="Update entire rsi data",
    status_code=status.HTTP_200_OK
)
async def update_entire_rsi():
    Indicator.update_entire_rsi()


@router.get(
    "/update_entire_macd",
    name="Update entire macd data",
    status_code=status.HTTP_200_OK
)
async def update_entire_macd():
    Indicator.update_entire_macd()


@router.get(
    "/vn100",
    name="Get VN100 symbols",
    status_code=status.HTTP_200_OK
)
async def get_vn100(payload: IndexPayload) -> dict:
    return Indicator.get_vn100(payload.source)


@router.post(
    "/get_price",
    name="Get price",
    status_code=status.HTTP_200_OK
)
async def get_price_data(payload: PricePayload) -> Sequence[dict]:
    price_df = Indicator.get_price(payload.symbol,
                                   payload.count_back,
                                   payload.source)
    if price_df is None:
        return [{"message": "Error"}]
    return json.loads(price_df.to_json(orient="records"))


@router.post(
    "/get_rsi",
    name="Get data of RSI",
    status_code=status.HTTP_200_OK
)
async def get_rsi_data(payload: RSIPayload) -> Sequence[dict]:
    rsi_df = Indicator.get_rsi(
        payload.symbol,
        payload.periods,
        payload.smooth_k,
        payload.smooth_d
    )
    if rsi_df is None:
        return [{"message": "Error"}]
    return json.loads(rsi_df.to_json(orient="records"))


@router.post(
    "/get_macd",
    name="Get data of MACD",
    status_code=status.HTTP_200_OK
)
async def get_macd_data(payload: MACDPayload) -> Sequence[dict]:
    macd_df = Indicator.get_macd(
        payload.symbol
    )
    if macd_df is None:
        return [{"message": "Error"}]
    return json.loads(macd_df.to_json(orient="records"))


@router.post(
    "/get_ichimoku",
    name="Get data of Ichimoku cloud",
    status_code=status.HTTP_200_OK
)
async def get_ichimoku_data(payload: IchimokuPayload) -> Sequence[dict]:
    price_df = Indicator.get_price(payload.symbol)
    ichimoku_df = Indicator.get_ichimoku_cloud(price_df,
                                               payload.conversion_period,
                                               payload.base_period,
                                               payload.span_b_period,
                                               payload.displacement)
    if ichimoku_df is None:
        return [{"message": "Error"}]
    return json.loads(ichimoku_df.to_json(orient="records"))