add sma & market cap for stock alert report

.gitignore

@@ -148,3 +148,4 @@ exclude_coins.json
 *.db-shm
 *.db-wal
 configs.json
+.DS_Store

smrti_quant_alerts/alerts/stock_alerts/stock_alert.py

@@ -62,14 +62,18 @@ def send_stocks_info_as_csv(self, is_newly_added_dict: Dict[StockSymbol, bool],
                   "Founded Year/IPO Date", "Is SP500", "Is Nasdaq", "Is NYSE",
                   "Is Newly Added", "Timeframe Alerted", "Free Cash Flow", "Net Income",
                   "Free Cash Flow Margin", "Revenue 1Y CAGR", "Revenue 3Y CAGR",
-                  "Revenue 5Y CAGR", "One Day Volume"]
+                  "Revenue 5Y CAGR", "One Day Volume", "Market Cap", "Close > 4H SMA200", "Close > 1D SMA200"]
         stocks = sorted(is_newly_added_dict.keys(), key=lambda x: x.ticker)
         stock_stats = self.get_semi_year_stocks_stats(stocks)
         stock_revenue_cagr = self.get_stocks_revenue_cagr(stocks)
 
         for k, v in stock_stats.items():
             stock_stats[k].update(stock_revenue_cagr[k])
 
+        stock_sma_data = self.get_close_price_sma_status(stocks, [200, 200], ["4hour", "1day"])
+
+        market_caps = self.get_stocks_market_cap(stocks)
+
         stock_timeframes = defaultdict(list)
         for timeframe, timeframe_stocks in timeframe_stocks_dict.items():
             for stock in timeframe_stocks:
@@ -81,7 +85,8 @@ def send_stocks_info_as_csv(self, is_newly_added_dict: Dict[StockSymbol, bool],
                        stock_stats[stock]["free_cash_flow"], stock_stats[stock]["net_income"],
                        stock_stats[stock]["free_cash_flow_margin"], stock_revenue_cagr[stock]["revenue_1y_cagr"],
                        stock_revenue_cagr[stock]["revenue_3y_cagr"], stock_revenue_cagr[stock]["revenue_5y_cagr"],
-                       self._daily_volume[stock]] for stock in stocks]
+                       self._daily_volume[stock], market_caps.get(stock, 0), stock_sma_data[stock]["4hour"],
+                       stock_sma_data[stock]["1day"]] for stock in stocks]
 
         self._tg_bot.send_data_as_csv_file(csv_file_name, headers=header, data=stock_info)
         pdf_files = []
@@ -233,9 +238,9 @@ def run(self) -> None:
 
 if __name__ == '__main__':
     start = time.time()
-    stock_alert = StockAlert("stock_alert", tg_type="TEST", email=True,
+    stock_alert = StockAlert("stock_alert", tg_type="TEST", email=False,
                              timeframe_list=["1m"],
                              #timeframe_list=["1m", "3m", "6m", "1y", "3y", "5y", "10y"],
-                             ai_analysis=True, daily_volume_threshold=500000)
+                             ai_analysis=False, daily_volume_threshold=500000)
     stock_alert.run()
     print(f"Time taken: {round(time.time() - start, 2)} seconds")

smrti_quant_alerts/stock_crypto_api/stock_api.py

@@ -224,6 +224,51 @@ def get_top_market_cap_stocks(self, top_n: int = 100) -> List[List[Union[Decimal
 
         return res[:top_n]
 
+    @error_handling("financialmodelingprep", default_val={})
+    def get_close_price_sma_status(self, stock_list: List[StockSymbol], num_of_days_list: List[int],
+                                   timeframes: List[str]) -> Dict[StockSymbol, Dict[str, bool]]:
+        """
+        Get the mapping for whether the close price is higher than sma
+        :param stock_list: [StockSymbol, ...]
+        :param num_of_days_list: [int, ...], e.g. [200, 400] for sma200, sma400
+        :param timeframes: [str, ...], i.e. ["1min", "5min", "15min", "30min", "1hour", "4hour", "1day"]
+        :return: {StockSymbol: {"1min": True, "5min": False, "15min": True, "30min": False}}
+        """
+        res = defaultdict(dict)
+        for stock in stock_list:
+            close_price_url = f"https://financialmodelingprep.com/api/v3/quote-short/{stock}" \
+                              f"?apikey={self.FMP_API_KEY}"
+            response = requests.get(close_price_url, timeout=10)
+            response = response.json()[0]
+            close_price = response.get("price", 0)
+            for num_of_day, timeframe in zip(num_of_days_list, timeframes):
+                from_day = get_datetime_now() - datetime.timedelta(days=2)
+                sma_url = f"https://financialmodelingprep.com/api/v3/technical_indicator/{timeframe}/{stock}" \
+                          f"?type=sma&period={num_of_day}&apikey={self.FMP_API_KEY}" \
+                          f"&from={from_day.strftime('%Y-%m-%d')}"
+                response = requests.get(sma_url, timeout=10)
+                response = response.json()[0]
+                sma = response.get("sma", 0)
+
+                res[stock][timeframe] = close_price > sma
+        return res
+
+    @error_handling("financialmodelingprep", default_val={})
+    def get_stocks_market_cap(self, stock_list: List[StockSymbol]) -> Dict[StockSymbol, Decimal]:
+        """
+        Get stock market cap
+
+        :param stock_list: [StockSymbol, ...]
+        :return: {StockSymbol: market_cap}
+        """
+        res = defaultdict(Decimal)
+        for stock in stock_list:
+            api_url = f"{self.FMP_API_URL}market-capitalization/{stock.ticker}?apikey={self.FMP_API_KEY}"
+            response = requests.get(api_url, timeout=10).json()
+            if response:
+                res[stock] = Decimal(response[0].get("marketCap", 0))
+        return res
+
     def get_semi_year_stocks_stats(self, stock_list: List[StockSymbol]) -> Dict[StockSymbol, Dict[str, str]]:
         """
         Get stock free cash flow, net income, free cash flow margin
@@ -310,6 +355,7 @@ def get_stock_revenue_cagr(stock: StockSymbol) -> None:
 
 
 if __name__ == "__main__":
+    stock = StockSymbol("AAOI")
     stock_api = StockApi()
-    revenue_cagr = stock_api.get_stocks_revenue_cagr(stock_list=[StockSymbol("CING")])
-    print(revenue_cagr)
+    market_cap = stock_api.get_stocks_market_cap([stock])
+    print(market_cap)