feat: update price and rsi data of vn100

models/ichimoku.py

@@ -1,17 +1,9 @@
-from pydantic import BaseModel, root_validator
+from pydantic import BaseModel
 
 
 class IchimokuPayload(BaseModel):
     symbol: str
-    count_back: int = 200
-    symbol_type: str = "stock"
     conversion_period: int = 9
     base_period: int = 26
     span_b_period: int = 52
     displacement: int = 26
-
-    @root_validator
-    def validate_dates(cls, values):
-        if values['symbol_type'] not in ["index", "stock"]:
-            raise ValueError('type should be index or stock')
-        return values

models/price.py

@@ -1,13 +1,5 @@
-from pydantic import BaseModel, root_validator
+from pydantic import BaseModel
 
 
 class PricePayload(BaseModel):
     symbol: str
-    count_back: int = 200
-    symbol_type: str = "stock"
-
-    @root_validator
-    def validate_dates(cls, values):
-        if values['symbol_type'] not in ["index", "stock"]:
-            raise ValueError('type should be index or stock')
-        return values

routes/data.py

@@ -10,17 +10,39 @@ router = APIRouter()
 
 
 @router.get(
-    "/update_rsi",
-    name="Update RSI data",
+    "/update_daily_price",
+    name="Update daily price data",
     status_code=status.HTTP_200_OK
 )
-async def update_rsi_date():
-    Indicator.get_rsi(
-        symbol="FPT",
-        periods=14,
-        smooth_k=3,
-        smooth_d=3
-    )
+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.post(
@@ -29,9 +51,7 @@ async def update_rsi_date():
     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.symbol_type)
+    price_df = Indicator.get_price(payload.symbol)
     if price_df is None:
         return [{"message": "Error"}]
     return json.loads(price_df.to_json(orient="records"))
@@ -60,9 +80,7 @@ async def get_rsi_data(payload: RSIPayload) -> Sequence[dict]:
     status_code=status.HTTP_200_OK
 )
 async def get_ichimoku_data(payload: IchimokuPayload) -> Sequence[dict]:
-    price_df = Indicator.get_price(payload.symbol,
-                                   payload.count_back,
-                                   payload.symbol_type)
+    price_df = Indicator.get_price(payload.symbol)
     ichimoku_df = Indicator.get_ichimoku_cloud(price_df,
                                                payload.conversion_period,
                                                payload.base_period,

services/indicator.py

@@ -2,9 +2,9 @@ import requests
 from datetime import datetime, timedelta
 import pandas as pd
 import numpy as np
-from vnstock import longterm_ohlc_data, stock_historical_data, price_board
+from vnstock import longterm_ohlc_data
 from pymongo import MongoClient, ASCENDING, DESCENDING
-from utils.config import DATE_FORMAT
+from utils.config import DATE_FORMAT, VN100_URL
 from .utils import Utility
 import os
 # from dotenv import load_dotenv
@@ -20,6 +20,9 @@ INDICATORS = {
     "Williams %R": "GetWilliamChart",
     "BollingerBand": "GetBollingerBandChart",
 }
+updating_price = False
+updating_rsi = False
+updating_macd = False
 
 
 class Indicator:
@@ -27,23 +30,245 @@ class Indicator:
         pass
 
     @staticmethod
-    def get_price(symbol: str,
-                  count_back: int = 200,
-                  symbol_type: str = "stock",) -> pd.DataFrame:
-        resolution = "D"
-        end_date = datetime.now()
-        delta = timedelta(days=count_back)
-        start_date = end_date - delta
-        start_date = datetime.strftime(start_date, '%Y-%m-%d')
-        end_date = datetime.strftime(end_date, '%Y-%m-%d')
-        df = longterm_ohlc_data(symbol, start_date, end_date,
-                                resolution, symbol_type).reset_index(drop=True)
-        df[['open', 'high', 'low', 'close']] = \
-            df[['open', 'high', 'low', 'close']] * 1000
-        # convert open, high, low, close to int
-        df[['open', 'high', 'low', 'close']] = \
-            df[['open', 'high', 'low', 'close']].astype(int)
-        return df
+    def update_daily_price() -> None:
+        global updating_price
+        if updating_price:
+            return
+        updating_price = True
+        datetime_now = datetime.utcnow()
+        hour = datetime_now.hour
+        weekday = datetime_now.weekday()
+        start_date = datetime_now.date()
+        delta = timedelta(days=1)
+        end_date = start_date + delta
+        start_date = start_date.strftime(DATE_FORMAT)
+        end_date = end_date.strftime(DATE_FORMAT)
+        try:
+            if weekday < 5 and hour >= 12:
+                uri = os.environ.get("MONGODB_URI")
+                client = MongoClient(uri)
+                database = client.get_database("data")
+                collection = database.get_collection("price")
+                tmp_df = longterm_ohlc_data("MBB",
+                                            start_date,
+                                            end_date,
+                                            "D",
+                                            "stock").reset_index(drop=True)
+                if tmp_df.shape[0] == 0:
+                    updating_price = False
+                    return
+                last_date = str(tmp_df["time"])
+                newest_record = \
+                    collection.find_one(sort=[("_id", DESCENDING)])
+                if newest_record["time"] != last_date:
+                    lst_symbols = list(newest_record["value"].keys())
+                    record = {}
+                    record["time"] = last_date
+                    values = []
+                    for symbol in lst_symbols:
+                        df = longterm_ohlc_data(symbol,
+                                                start_date,
+                                                end_date,
+                                                resolution="D",
+                                                type="stock"
+                                                ).reset_index(drop=True)
+                        df[["open", "high", "low", "close"]] = \
+                            df[["open", "high", "low", "close"]] * 1000
+                        # convert open, high, low, close to int
+                        df[["open", "high", "low", "close"]] = \
+                            df[["open", "high", "low", "close"]].astype(int)
+                        df = df[["ticker", "open",
+                                "high", "low", "close"]].iloc[-1]
+                        values.append(df.to_dict())
+                    record["value"] = values
+                collection.insert_one(record)
+                collection.find_one_and_delete({}, sort=[("_id", ASCENDING)])
+                print("Updated price")
+                updating_price = False
+        except Exception as e:
+            print(f"Error: {e}")
+            updating_price = False
+
+    @staticmethod
+    def update_entire_price() -> None:
+        global updating_price
+        if updating_price:
+            return
+        updating_price = True
+        datetime_now = datetime.utcnow()
+        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())
+            if weekday < 5 and hour < 12:
+                return
+            end_date = datetime_now.date()
+            delta = timedelta(days=300)
+            start_date = end_date - delta
+            start_date = start_date.strftime(DATE_FORMAT)
+            end_date = end_date.strftime(DATE_FORMAT)
+            lst_time = []
+            lst_values = []
+            if len(lst_symbols) != 100:
+                return
+            for symbol in lst_symbols:
+                df = longterm_ohlc_data(symbol,
+                                        start_date,
+                                        end_date,
+                                        resolution="D",
+                                        type="stock"
+                                        ).reset_index(drop=True).tail(150)
+                lst_time = list(df["time"])
+                df[["open", "high", "low", "close"]] = \
+                    df[["open", "high", "low", "close"]] * 1000
+                # convert open, high, low, close to int
+                df[["open", "high", "low", "close"]] = \
+                    df[["open", "high", "low", "close"]].astype(int)
+                df = df[["ticker", "open", "high", "low", "close"]]
+                lst_values.append(df.to_dict(orient="records"))
+            records = []
+            for i in range(len(lst_time)):
+                record = {}
+                record["time"] = lst_time[i]
+                values = []
+                for symbol_index in range(100):
+                    values.append(lst_values[symbol_index][i])
+                record["value"] = values
+                records.append(record)
+            uri = os.environ.get("MONGODB_URI")
+            client = MongoClient(uri)
+            database = client.get_database("data")
+            collection = database.get_collection("price")
+            if "price" in database.list_collection_names():
+                collection.drop()
+            collection.insert_many(records)
+            print("Updated entire price")
+            updating_price = False
+        except Exception as e:
+            print(f"Error: {e}")
+            updating_price = False
+
+    @staticmethod
+    def get_price(symbol) -> pd.DataFrame:
+        try:
+            symbol = symbol.upper()
+            uri = os.environ.get("MONGODB_URI")
+            client = MongoClient(uri)
+            database = client.get_database("data")
+            collection = database.get_collection("price")
+            result = list(collection.find())
+            tmp_df = pd.DataFrame(result[0]["value"])
+            lst_symbols = tmp_df["ticker"].values
+            if symbol not in lst_symbols:
+                return None
+            symbol_index = np.argwhere(lst_symbols == symbol)[0][0]
+            lst_values = []
+            for record in result:
+                value = record["value"][symbol_index]
+                value["time"] = record["time"]
+                lst_values.append(value)
+            return pd.DataFrame(lst_values)
+        except Exception:
+            return None
+
+    @staticmethod
+    def update_daily_rsi() -> None:
+        global updating_rsi
+        if updating_rsi:
+            return
+        updating_rsi = True
+        datetime_now = datetime.utcnow()
+        hour = datetime_now.hour
+        weekday = datetime_now.weekday()
+        start_date = datetime_now.date()
+        delta = timedelta(days=1)
+        end_date = start_date + delta
+        start_date = start_date.strftime(DATE_FORMAT)
+        end_date = end_date.strftime(DATE_FORMAT)
+        try:
+            if weekday < 5 and hour >= 12:
+                uri = os.environ.get("MONGODB_URI")
+                client = MongoClient(uri)
+                database = client.get_database("data")
+                collection = database.get_collection("price")
+                tmp_df = longterm_ohlc_data("MBB",
+                                            start_date,
+                                            end_date,
+                                            "D",
+                                            "stock").reset_index(drop=True)
+                if tmp_df.shape[0] == 0:
+                    return
+                last_date = str(tmp_df["time"])
+                newest_record = \
+                    collection.find_one(sort=[("_id", DESCENDING)])
+                if newest_record["time"] != last_date:
+                    lst_symbols = list(newest_record["value"].keys())
+                    record = {}
+                    record["time"] = last_date
+                    str_symbols = ','.join(lst_symbols)
+                    data = requests.get('https://apipubaws.tcbs.com.vn/stock-insight/v1/stock/second-tc-price?tickers={}'.format(str_symbols)).json()
+                    for i in data["data"]:
+                        record[i["t"]] = i["rsi"]
+                collection.insert_one(record)
+                collection.find_one_and_delete({}, sort=[("_id", ASCENDING)])
+                print("Updated daily rsi")
+                updating_rsi = False
+        except Exception as e:
+            print(f"Error: {e}")
+            updating_rsi = False
+
+    @staticmethod
+    def update_entire_rsi() -> None:
+        global updating_rsi
+        if updating_rsi:
+            return
+        updating_rsi = True
+        datetime_now = datetime.utcnow()
+        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())
+            if weekday < 5 and hour < 12:
+                return
+            get_time = True
+            lst_values = {}
+            if len(lst_symbols) != 100:
+                return
+            cnt = 0
+            for symbol in lst_symbols:
+                cnt += 1
+                url = PREFIX + INDICATORS["RSI"] \
+                    + f"?stockSymbol={symbol}&rangeSelector=3&periods=14"
+                data = requests.get(url).json()
+                rsi_df = pd.DataFrame(data["SeriesColection"][0]["Points"])
+                if get_time:
+                    lst_values["time"] = \
+                        rsi_df["Time"].apply(lambda x:
+                                             Utility.ts_to_date(x/1000))
+                    get_time = False
+                lst_values[symbol] = rsi_df["Value"].apply(lambda x: x[0])
+            df = pd.DataFrame(lst_values)
+            records = df.to_dict(orient="records")
+            uri = os.environ.get("MONGODB_URI")
+            client = MongoClient(uri)
+            database = client.get_database("data")
+            collection = database.get_collection("rsi")
+            if "rsi" in database.list_collection_names():
+                collection.drop()
+            collection.insert_many(records)
+            print("Updated entire RSI")
+            updating_rsi = False
+        except Exception as e:
+            print(f"Error: {e}")
+            updating_rsi = False
 
     @staticmethod
     def get_rsi(
@@ -58,45 +283,13 @@ class Indicator:
             client = MongoClient(uri)
             database = client.get_database("data")
             collection = database.get_collection("rsi")
-            datetime_now = datetime.utcnow()
-            hour = datetime_now.hour
-            weekday = datetime_now.weekday()
-            error = False
-            if weekday < 5 and hour >= 12:  # Check if data is updated
-                newest_record = collection.find_one(sort=[("_id", DESCENDING)])
-                delta = timedelta(days=20)
-                start_date = datetime_now - delta
-                start_date = start_date.strftime(DATE_FORMAT)
-                end_date = datetime_now.strftime(DATE_FORMAT)
-                tmp_df = stock_historical_data(symbol, start_date, end_date)
-                last_date = str(tmp_df["time"].iloc[-1])
-                if newest_record["time"] != last_date:
-                    try:
-                        lst_symbols = list(newest_record.keys())[2:]
-                        record = {}
-                        record["time"] = last_date
-                        for s in lst_symbols:
-                            url = PREFIX + INDICATORS["RSI"] + f"?stockSymbol={s}&rangeSelector=2&periods={periods}"
-                            data = requests.get(url).json()
-                            lst_rsi = data["SeriesColection"][0]["Points"]
-                            record[s] = lst_rsi[-1]["Value"][0]
-                        collection.find_one_and_delete({}, sort=[("_id", ASCENDING)])
-                        collection.insert_one(record)
-                        print("Updated data")
-                    except Exception:
-                        error = True
             try:
                 records = list(collection.find())
             except Exception as e:
                 print(f"Error: {e}")
-            print("load successfully")
             record_df = pd.DataFrame(records).drop(columns=["_id"])
             record_df = \
                 record_df[["time", symbol]].rename(columns={symbol: "rsi"})
-            if error:
-                new_df = price_board(symbol)
-                value = [last_date, new_df["RSI"][0]]
-                record_df.loc[len(record_df)] = value
             record_df["stoch_rsi"] = \
                 Indicator.stoch_rsi(record_df["rsi"], periods)
             record_df["stoch_rsi_smooth_k"] = \

utils/backup.py

@@ -1,11 +0,0 @@
-from pymongo.mongo_client import MongoClient
-from pymongo.server_api import ServerApi
-uri = "mongodb+srv://dacap3h:vMRoeFxX7Jjn2O5m@backup-data.m6pxu2j.mongodb.net/"
-# Create a new client and connect to the server
-client = MongoClient(uri, server_api=ServerApi('1'))
-# Send a ping to confirm a successful connection
-try:
-    client.admin.command('ping')
-    print("Pinged your deployment. You successfully connected to MongoDB!")
-except Exception as e:
-    print(e)

utils/config.py

@@ -10,3 +10,4 @@ tcbs_headers = {
   'Referer': 'https://tcinvest.tcbs.com.vn/',
   'sec-ch-ua-platform': '"Windows"'
 }
+VN100_URL = "https://www.hsx.vn/Modules/Listed/Web/StockIndex/188803177?_search=false&nd=1707328622054&rows=2147483647&page=1&sidx=id&sord=desc"