Spaces:
Running
Running
File size: 3,329 Bytes
884dad1 8315578 be1eeae 884dad1 90e3571 e565da6 90e3571 e565da6 90e3571 8315578 be1eeae 884dad1 5c36aa4 06613fc 9da4532 884dad1 06613fc 9da4532 884dad1 8315578 884dad1 e565da6 048da70 06613fc 9da4532 884dad1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 |
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"))
|