IntelliStock/data-api
0
1import requests2from datetime import datetime, timedelta3import pandas as pd4import numpy as np5from vnstock import longterm_ohlc_data6from pymongo import MongoClient, ASCENDING, DESCENDING7from utils.config import DATE_FORMAT, VN100_URL8from .utils import Utility9import os10# from dotenv import load_dotenv11# load_dotenv()12 13PREFIX = "https://www.hsx.vn/Modules/Chart/StaticChart/"14TIME = {"1m": 2, "3m": 3, "6m": 4, "1y": 5, "2y": 6, "5y": 7}15INDICATORS = {16 "EMA": "GetEmaChart",17 "MACD": "GetMacdChart",18 "RSI": "GetRsiChart",19 "Momentum": "GetMomentumChart",20 "Williams %R": "GetWilliamChart",21 "BollingerBand": "GetBollingerBandChart",22}23updating_price = False24updating_rsi = False25updating_macd = False26 27 28class Indicator:29 def __init__(self) -> None:30 pass31 32 @staticmethod33 def get_vn100(source: str = "database") -> dict:34 try:35 lst_symbols = []36 if source == "database":37 uri = os.environ.get("MONGODB_URI")38 client = MongoClient(uri)39 database = client.get_database("data")40 collection = database.get_collection("rsi")41 newest_record = collection.find_one(sort=[("_id", ASCENDING)])42 lst_symbols = list(newest_record.keys())[2:]43 elif source == "hsx":44 data = requests.get(VN100_URL).json()45 data = data["rows"]46 for value in data:47 lst_symbols.append(value["cell"][2].strip())48 return {"symbols": lst_symbols}49 except Exception as e:50 return {"message": f"Caught error {e}."}51 52 @staticmethod53 async def update_daily_price() -> dict:54 global updating_price55 if updating_price:56 print("The data has been updating.")57 return58 updating_price = True59 datetime_now = datetime.utcnow()60 hour = datetime_now.hour61 weekday = datetime_now.weekday()62 end_date = datetime_now.date()63 delta = timedelta(days=5)64 start_date = end_date - delta65 start_date = start_date.strftime(DATE_FORMAT)66 end_date = end_date.strftime(DATE_FORMAT)67 try:68 if weekday < 5 and hour >= 12:69 uri = os.environ.get("MONGODB_URI")70 client = MongoClient(uri)71 database = client.get_database("data")72 collection = database.get_collection("price")73 tmp_df = longterm_ohlc_data("FPT",74 start_date,75 end_date,76 "D",77 "stock").reset_index(drop=True)78 if tmp_df.shape[0] == 0:79 updating_price = False80 print("There is no data to updated")81 return82 tmp_df = tmp_df.iloc[-1]83 last_date = str(tmp_df["time"])84 newest_record = \85 collection.find_one(sort=[("_id", DESCENDING)])86 ret = None87 if newest_record["time"] != last_date:88 rsi_collection = database.get_collection("rsi")89 tmp_record = \90 rsi_collection.find_one(sort=[("_id", DESCENDING)])91 lst_symbols = list(tmp_record.keys())[2:]92 record = {}93 record["time"] = last_date94 values = []95 for symbol in lst_symbols:96 df = longterm_ohlc_data(symbol,97 start_date,98 end_date,99 resolution="D",100 type="stock"101 ).reset_index(drop=True)102 df[["open", "high", "low", "close"]] = \103 df[["open", "high", "low", "close"]] * 1000104 # convert open, high, low, close to int105 df[["open", "high", "low", "close"]] = \106 df[["open", "high", "low", "close"]].astype(int)107 df = df[["ticker", "open",108 "high", "low", "close", "volume"]].iloc[-1]109 values.append(df.to_dict())110 record["value"] = values111 collection.insert_one(record)112 ret = "Updated data."113 else:114 ret = "The data is up to date."115 updating_price = False116 print(ret)117 except Exception as e:118 updating_price = False119 print(f"Caught error {e}.")120 121 @staticmethod122 def update_entire_price() -> None:123 global updating_price124 if updating_price:125 return {"message": "The data has been updating."}126 updating_price = True127 datetime_now = datetime.utcnow()128 hour = datetime_now.hour129 weekday = datetime_now.weekday()130 try:131 symbol_dict = Indicator.get_vn100(source="hsx")132 lst_symbols = symbol_dict["symbols"]133 if weekday < 5 and hour < 12:134 updating_price = False135 return {"message": "The market is in trading."}136 end_date = datetime_now.date()137 delta = timedelta(days=350)138 start_date = end_date - delta139 start_date = start_date.strftime(DATE_FORMAT)140 end_date = end_date.strftime(DATE_FORMAT)141 lst_time = []142 lst_values = []143 if len(lst_symbols) != 100:144 updating_price = False145 return {"message": "Not enough 100 symbols."}146 for symbol in lst_symbols:147 df = longterm_ohlc_data(symbol,148 start_date,149 end_date,150 resolution="D",151 type="stock"152 ).reset_index(drop=True).tail(205)153 lst_time = list(df["time"])154 df[["open", "high", "low", "close"]] = \155 df[["open", "high", "low", "close"]] * 1000156 # convert open, high, low, close to int157 df[["open", "high", "low", "close"]] = \158 df[["open", "high", "low", "close"]].astype(int)159 df = df[["ticker", "open", "high", "low", "close", "volume"]]160 lst_values.append(df.to_dict(orient="records"))161 records = []162 for i in range(len(lst_time)):163 record = {}164 record["time"] = lst_time[i]165 values = []166 for symbol_index in range(100):167 values.append(lst_values[symbol_index][i])168 record["value"] = values169 records.append(record)170 uri = os.environ.get("MONGODB_URI")171 client = MongoClient(uri)172 database = client.get_database("data")173 collection = database.get_collection("price")174 if "price" in database.list_collection_names():175 collection.drop()176 collection.insert_many(records)177 updating_price = False178 return {"message": "Updated data."}179 except Exception as e:180 updating_price = False181 return {"message": f"Caught error {e}."}182 183 @staticmethod184 def get_price(symbol: str,185 count_back: int = 150,186 source: str = "database") -> pd.DataFrame:187 try:188 symbol = symbol.upper()189 if source == "database":190 uri = os.environ.get("MONGODB_URI")191 client = MongoClient(uri)192 database = client.get_database("data")193 collection = database.get_collection("price")194 result = list(collection.find())195 tmp_df = pd.DataFrame(result[0]["value"])196 lst_symbols = tmp_df["ticker"].values197 if symbol not in lst_symbols:198 return pd.DataFrame(199 [{"message": "The symbol is not existed"}]200 )201 symbol_index = np.argwhere(lst_symbols == symbol)[0][0]202 lst_values = []203 for record in result:204 value = record["value"][symbol_index]205 value["time"] = record["time"]206 lst_values.append(value)207 return pd.DataFrame(lst_values)208 elif source == "vnstock":209 datetime_now = datetime.utcnow()210 end_date = datetime_now.date()211 delta = timedelta(days=count_back)212 start_date = end_date - delta213 start_date = start_date.strftime(DATE_FORMAT)214 end_date = end_date.strftime(DATE_FORMAT)215 df = longterm_ohlc_data(symbol,216 start_date,217 end_date,218 resolution="D",219 type="stock"220 ).reset_index(drop=True)221 df[["open", "high", "low", "close"]] = \222 df[["open", "high", "low", "close"]] * 1000223 # convert open, high, low, close to int224 df[["open", "high", "low", "close"]] = \225 df[["open", "high", "low", "close"]].astype(int)226 return df[["time", "ticker", "open", "high", "low", "close"]]227 except Exception as e:228 return pd.DataFrame(229 [{"message": f"Caught error {e}."}]230 )231 232 @staticmethod233 async def update_daily_rsi() -> None:234 global updating_rsi235 if updating_rsi:236 print("The data has been updating.")237 return238 # return {"message": "The data has been updating."}239 updating_rsi = True240 datetime_now = datetime.utcnow()241 hour = datetime_now.hour242 weekday = datetime_now.weekday()243 end_date = datetime_now.date()244 delta = timedelta(days=5)245 start_date = end_date - delta246 start_date = start_date.strftime(DATE_FORMAT)247 end_date = end_date.strftime(DATE_FORMAT)248 try:249 ret = None250 if weekday < 5 and hour >= 12:251 uri = os.environ.get("MONGODB_URI")252 client = MongoClient(uri)253 database = client.get_database("data")254 collection = database.get_collection("rsi")255 tmp_df = longterm_ohlc_data("FPT",256 start_date,257 end_date,258 "D",259 "stock").reset_index(drop=True)260 if tmp_df.shape[0] == 0:261 updating_rsi = False262 print("The market is in trading.")263 return264 # return {"message": "The market is in trading."}265 tmp_df = tmp_df.iloc[-1]266 last_date = str(tmp_df["time"])267 newest_record = \268 collection.find_one(sort=[("_id", DESCENDING)])269 if newest_record["time"] != last_date:270 lst_symbols = list(newest_record.keys())[2:]271 record = {}272 record["time"] = last_date273 for batch_idx in range(10):274 str_symbols = ','.join(275 lst_symbols[batch_idx*10:(batch_idx+1)*10])276 data = requests.get('https://apipubaws.tcbs.com.vn/stock-insight/v1/stock/second-tc-price?tickers={}'.format(str_symbols)).json()277 for i in data["data"]:278 record[i["t"]] = i["rsi"]279 collection.insert_one(record)280 ret = "Updated data."281 else:282 ret = "The data is up to date."283 else:284 ret = "The market is in trading."285 updating_rsi = False286 print(ret)287 except Exception as e:288 updating_rsi = False289 print(f"Caught error {e}.")290 291 @staticmethod292 def update_entire_rsi() -> None:293 global updating_rsi294 if updating_rsi:295 return {"message": "The data has been updating."}296 updating_rsi = True297 datetime_now = datetime.utcnow()298 hour = datetime_now.hour299 weekday = datetime_now.weekday()300 try:301 symbol_dict = Indicator.get_vn100(source="hsx")302 lst_symbols = symbol_dict["symbols"]303 if weekday < 5 and hour < 12:304 updating_rsi = False305 return {"message": "The market is in trading."}306 get_time = True307 lst_values = {}308 if len(lst_symbols) != 100:309 updating_rsi = False310 return {"message": "Not enough 100 symbols."}311 cnt = 0312 for symbol in lst_symbols:313 cnt += 1314 url = PREFIX + INDICATORS["RSI"] \315 + f"?stockSymbol={symbol}&rangeSelector=3&periods=14"316 data = requests.get(url).json()317 rsi_df = pd.DataFrame(data["SeriesColection"][0]["Points"])318 if get_time:319 lst_values["time"] = \320 rsi_df["Time"].apply(lambda x:321 Utility.ts_to_date(x/1000))322 get_time = False323 lst_values[symbol] = rsi_df["Value"].apply(lambda x: x[0])324 df = pd.DataFrame(lst_values)325 records = df.to_dict(orient="records")326 uri = os.environ.get("MONGODB_URI")327 client = MongoClient(uri)328 database = client.get_database("data")329 collection = database.get_collection("rsi")330 if "rsi" in database.list_collection_names():331 collection.drop()332 collection.insert_many(records)333 updating_rsi = False334 return {"message": "Updated data."}335 except Exception as e:336 updating_rsi = False337 return {"message": f"Caught error {e}."}338 339 @staticmethod340 def get_rsi(341 symbol: str,342 periods: int = 14,343 smooth_k: int = 3,344 smooth_d: int = 3,345 ) -> pd.DataFrame:346 try:347 symbol = symbol.upper()348 uri = os.environ.get("MONGODB_URI")349 client = MongoClient(uri)350 database = client.get_database("data")351 collection = database.get_collection("rsi")352 records = list(collection.find())353 record_df = pd.DataFrame(records).drop(columns=["_id"])354 record_df = \355 record_df[["time", symbol]].rename(columns={symbol: "rsi"})356 record_df["stoch_rsi"] = \357 Indicator.stoch_rsi(record_df["rsi"], periods)358 record_df["stoch_rsi_smooth_k"] = \359 Indicator.stoch_rsi_smooth_k(record_df["stoch_rsi"], smooth_k)360 record_df["stoch_rsi_smooth_d"] = Indicator.stoch_rsi_smooth_d(361 record_df["stoch_rsi_smooth_k"], smooth_d362 )363 return record_df364 except Exception as e:365 return pd.DataFrame(366 [{"message": f"Caught error {e}."}]367 )368 369 @staticmethod370 def stoch_rsi(rsi: pd.Series, periods: int = 14) -> pd.Series:371 ma, mi = (372 rsi.rolling(window=periods).max(),373 rsi.rolling(window=periods).min(),374 )375 return (rsi - mi) * 100 / (ma - mi)376 377 @staticmethod378 def stoch_rsi_smooth_k(stoch_rsi: pd.Series, k: int) -> pd.Series:379 return stoch_rsi.rolling(window=k).mean()380 381 @staticmethod382 def stoch_rsi_smooth_d(stoch_rsi_k: pd.Series, d: int) -> pd.Series:383 return stoch_rsi_k.rolling(window=d).mean()384 385 @staticmethod386 async def update_entire_macd() -> None:387 global updating_macd388 if updating_macd:389 print("The data has been updating.")390 return391 updating_macd = True392 datetime_now = datetime.utcnow()393 hour = datetime_now.hour394 weekday = datetime_now.weekday()395 try:396 symbol_dict = Indicator.get_vn100(source="hsx")397 lst_symbols = symbol_dict["symbols"]398 if weekday < 5 and hour < 12:399 updating_macd = False400 print("The market is in trading.")401 return402 get_time = True403 lst_values = {}404 if len(lst_symbols) != 100:405 updating_macd = False406 print("Not enough 100 symbols.")407 return408 cnt = 0409 for symbol in lst_symbols:410 cnt += 1411 url = PREFIX + INDICATORS["MACD"] \412 + f"?stockSymbol={symbol}&rangeSelector=4&fastPeriod=12&slowPeriod=26&signalPeriod=9"413 data = requests.get(url).json()414 rsi_df = pd.DataFrame(415 data["SeriesColection"][0]["Points"]).tail(20)416 if get_time:417 lst_values["time"] = \418 rsi_df["Time"].apply(lambda x:419 Utility.ts_to_date(x/1000))420 get_time = False421 lst_values[symbol] = rsi_df["Value"].apply(lambda x: int(x[0]))422 df = pd.DataFrame(lst_values)423 records = df.to_dict(orient="records")424 uri = os.environ.get("MONGODB_URI")425 client = MongoClient(uri)426 database = client.get_database("data")427 collection = database.get_collection("macd")428 if "macd" in database.list_collection_names():429 collection.drop()430 collection.insert_many(records)431 updating_macd = False432 print("Updated data.")433 except Exception as e:434 updating_macd = False435 print(f"Caught error {e}.")436 437 @staticmethod438 def get_macd(439 symbol: str,440 ) -> pd.DataFrame:441 try:442 symbol = symbol.upper()443 uri = os.environ.get("MONGODB_URI")444 client = MongoClient(uri)445 database = client.get_database("data")446 collection = database.get_collection("macd")447 records = list(collection.find())448 record_df = pd.DataFrame(records).drop(columns=["_id"])449 record_df = \450 record_df[["time", symbol]].rename(columns={symbol: "macd"})451 return record_df452 except Exception as e:453 return pd.DataFrame(454 [{"message": f"Caught error {e}."}]455 )456 457 @staticmethod458 def get_ichimoku_cloud(459 df: pd.DataFrame,460 conversion_period=9,461 base_period=26,462 span_b_period=52,463 displacement=26,464 ) -> pd.DataFrame:465 space_displacement = np.full(displacement, np.nan)466 tenkan_sen = (467 df["high"].rolling(window=conversion_period).max()468 + df["low"].rolling(window=conversion_period).min()469 ) / 2470 kijun_sen = (471 df["high"].rolling(window=base_period).max()472 + df["low"].rolling(window=base_period).min()473 ) / 2474 senkou_span_a = (tenkan_sen + kijun_sen) / 2475 senkou_span_b = (476 df["high"].rolling(window=span_b_period).max()477 + df["low"].rolling(window=span_b_period).min()478 ) / 2479 chikou_span = df["close"].shift(-displacement)480 481 last_date = datetime.strptime(df["time"].iloc[-1], DATE_FORMAT)482 lst_date = Utility.generate_dates(last_date, displacement)483 time = np.concatenate((df["time"], lst_date))484 tenkan_sen = np.concatenate((tenkan_sen, space_displacement))485 kijun_sen = np.concatenate((kijun_sen, space_displacement))486 senkou_span_a = np.concatenate((space_displacement, senkou_span_a))487 senkou_span_b = np.concatenate((space_displacement, senkou_span_b))488 chikou_span = np.concatenate((chikou_span, space_displacement))489 490 data_dict = {491 "time": time,492 "tenkan_sen": tenkan_sen,493 "kijun_sen": kijun_sen,494 "senkou_span_a": senkou_span_a,495 "senkou_span_b": senkou_span_b,496 "chikou_span": chikou_span,497 "tenkan_kijun": tenkan_sen - kijun_sen,498 "kumo_cloud": senkou_span_a - senkou_span_b499 }500 return pd.DataFrame(data_dict)501