Convert date to estimated date

services/indicator.py

@@ -5,6 +5,7 @@ import numpy as np
 from vnstock import longterm_ohlc_data, stock_historical_data, price_board
 from pymongo import MongoClient, ASCENDING, DESCENDING
 from utils.config import DATE_FORMAT
+from .utils import Utility
 import os
 # from dotenv import load_dotenv
 # load_dotenv()
@@ -27,7 +28,7 @@ class Indicator:
 
     @staticmethod
     def get_price(symbol: str,
-                  count_back: int = 100,
+                  count_back: int = 200,
                   symbol_type: str = "stock",) -> pd.DataFrame:
         resolution = "D"
         end_date = datetime.now()
@@ -51,58 +52,61 @@ class Indicator:
         smooth_k: int = 3,
         smooth_d: int = 3,
     ) -> pd.DataFrame:
-        symbol = symbol.upper()
-        uri = os.environ.get("MONGODB_URI")
-        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("MBB", 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"] = \
-            Indicator.stoch_rsi_smooth_k(record_df["stoch_rsi"], smooth_k)
-        record_df["stoch_rsi_smooth_d"] = Indicator.stoch_rsi_smooth_d(
-            record_df["stoch_rsi_smooth_k"], smooth_d
-        )
-        return record_df
+            symbol = symbol.upper()
+            uri = os.environ.get("MONGODB_URI")
+            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"] = \
+                Indicator.stoch_rsi_smooth_k(record_df["stoch_rsi"], smooth_k)
+            record_df["stoch_rsi_smooth_d"] = Indicator.stoch_rsi_smooth_d(
+                record_df["stoch_rsi_smooth_k"], smooth_d
+            )
+            return record_df
+        except Exception:
+            return None
 
     @staticmethod
     def stoch_rsi(rsi: pd.Series, periods: int = 14) -> pd.Series:
@@ -144,9 +148,9 @@ class Indicator:
         ) / 2
         chikou_span = df["close"].shift(-displacement)
 
-        date_displacement = np.array(
-            list(map(str, np.arange(1, displacement+1))))
-        time = np.concatenate((df["time"], date_displacement))
+        last_date = datetime.strptime(df["time"].iloc[-1], DATE_FORMAT)
+        lst_date = Utility.generate_dates(last_date, displacement)
+        time = np.concatenate((df["time"], lst_date))
         tenkan_sen = np.concatenate((tenkan_sen, space_displacement))
         kijun_sen = np.concatenate((kijun_sen, space_displacement))
         senkou_span_a = np.concatenate((space_displacement, senkou_span_a))

services/utils.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timedelta
 from utils.config import DATE_FORMAT
 
 
@@ -6,9 +6,22 @@ class Utility:
     def __init__(self) -> None: pass
 
     @staticmethod
-    def ts_to_date(ts, format=DATE_FORMAT) -> str:
+    def ts_to_date(ts: int, format=DATE_FORMAT) -> str:
         return datetime.fromtimestamp(ts).strftime(format)
 
     @staticmethod
-    def date_to_ts(date, format=DATE_FORMAT) -> int:
+    def date_to_ts(date: str, format=DATE_FORMAT) -> int:
         return int(datetime.timestamp(datetime.strptime(date, format)))
+
+    @staticmethod
+    def generate_dates(start_date: datetime.date,
+                       num_days: int,
+                       format=DATE_FORMAT):
+        lst_date = []
+        current_date = start_date
+        for i in range(num_days):
+            current_date += timedelta(days=1)
+            while current_date.weekday() >= 5:
+                current_date += timedelta(days=1)
+            lst_date.append(current_date.strftime(format))
+        return lst_date