feat: get vn100 symbols

models/index.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel, validator
+
+
+class IndexPayload(BaseModel):
+    source: str = "database"
+
+    @validator("source")
+    def valid_source(cls, v):
+        if v not in ["database", "hsx"]:
+            raise ValueError('source must be database or hsx')
+        return v

routes/data.py

@@ -3,6 +3,7 @@ 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
@@ -55,6 +56,15 @@ 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",

services/indicator.py

@@ -29,6 +29,26 @@ class Indicator:
     def __init__(self) -> None:
         pass
 
+    @staticmethod
+    def get_vn100(source: str = "database") -> dict:
+        try:
+            lst_symbols = []
+            if source == "database":
+                uri = os.environ.get("MONGODB_URI")
+                client = MongoClient(uri)
+                database = client.get_database("data")
+                collection = database.get_collection("rsi")
+                newest_record = collection.find_one()
+                lst_symbols = list(newest_record["value"].keys())
+            elif source == "hsx":
+                data = requests.get(VN100_URL).json()
+                data = data["rows"]
+                for value in data:
+                    lst_symbols.append(value["cell"][2].strip())
+            return {"symbols": lst_symbols}
+        except Exception as e:
+            return {"message": f"Caught error {e}"}
+
     @staticmethod
     def update_daily_price() -> None:
         global updating_price
@@ -99,11 +119,8 @@ class Indicator:
         hour = datetime_now.hour
         weekday = datetime_now.weekday()
         try:
-            data = requests.get(VN100_URL).json()
-            data = data["rows"]
-            lst_symbols = []
-            for value in data:
-                lst_symbols.append(value["cell"][2].strip())
+            symbol_dict = Indicator.get_vn100(source="hsx")
+            lst_symbols = symbol_dict["symbols"]
             if weekday < 5 and hour < 12:
                 updating_price = False
                 return
@@ -260,11 +277,8 @@ class Indicator:
         hour = datetime_now.hour
         weekday = datetime_now.weekday()
         try:
-            data = requests.get(VN100_URL).json()
-            data = data["rows"]
-            lst_symbols = []
-            for value in data:
-                lst_symbols.append(value["cell"][2].strip())
+            symbol_dict = Indicator.get_vn100(source="hsx")
+            lst_symbols = symbol_dict["symbols"]
             if weekday < 5 and hour < 12:
                 updating_rsi = False
                 return
@@ -357,11 +371,8 @@ class Indicator:
         hour = datetime_now.hour
         weekday = datetime_now.weekday()
         try:
-            data = requests.get(VN100_URL).json()
-            data = data["rows"]
-            lst_symbols = []
-            for value in data:
-                lst_symbols.append(value["cell"][2].strip())
+            symbol_dict = Indicator.get_vn100(source="hsx")
+            lst_symbols = symbol_dict["symbols"]
             if weekday < 5 and hour < 12:
                 updating_macd = False
                 return