floating_shares_alert

smrti_quant_alerts/alerts/stock_alerts/floating_shares_alert.py

@@ -1,8 +1,10 @@
 import logging
 import uuid
 import os
+import time
 import threading
-from typing import List
+from typing import List, Tuple, Dict, Union
+from collections import defaultdict
 
 import pandas as pd
 
@@ -49,7 +51,8 @@ def _load_stocks_info(self) -> None:
         stocks = self.get_nasdaq_list() + self.get_nyse_list()
         self.get_stock_info(stocks)
 
-    def _generate_floating_shares(self, symbols: List[StockSymbol]) -> str:
+    def _generate_floating_shares(self, symbols: List[StockSymbol]) \
+            -> Tuple[str, Dict[StockSymbol, List[Union[str, float]]]]:
         """
         Generate outstanding shares for the list of symbols
 
@@ -59,12 +62,13 @@ def _generate_floating_shares(self, symbols: List[StockSymbol]) -> str:
         """
         floating_shares = self.get_floating_shares(symbols)
         shares_content = ""
+
         for symbol, shares in floating_shares.items():
             if shares and len(shares) == 4:
                 has_increased = shares[1] > shares[3]
                 shares_content += f"~ {symbol}: {shares[3]} ({shares[2]}) -> {shares[1]} ({shares[0]}) " \
                                   f"{'+' if has_increased else '-'}\n\n"
-        return shares_content
+        return shares_content, floating_shares
 
     def _generate_file_8k_summary_csv(self) -> None:
         """
@@ -78,24 +82,53 @@ def _generate_file_8k_summary_csv(self) -> None:
         df = pd.DataFrame(data, columns=headers)
         df.to_csv(self._file_8k_summary_filename, index=False)
 
+    def _generate_floating_shares_summary_xlsx(
+            self, floating_shares: Dict[StockSymbol, List[Union[str, float]]]) -> None:
+        """
+        Generate the floating shares summary xlsx
+        """
+        self._load_stocks_info_thread.join()
+        industry_stock_floating_shares_mapping = defaultdict(list)
+        headers = ["stock", "previous date", "previous floating shares",
+                   "current date", "current floating shares", "has increased"]
+        for stock, floating_share in floating_shares.items():
+            industry_stock_floating_shares_mapping[stock.gics_sector].append(
+                [stock.ticker] + floating_share + [str(floating_share[1] > floating_share[3])]
+            )
+        with pd.ExcelWriter(self._floating_shares_summary_filename) as writer:
+            for industry, rows in industry_stock_floating_shares_mapping.items():
+                if industry == "":
+                    industry = "Others"
+                df = pd.DataFrame(rows, dtype=str)
+                df.to_excel(writer, sheet_name=industry, header=headers, index=False)
+
     def run(self) -> None:
         """
         Run the alert
         """
         self._generate_file_8k_summary_csv()
         # email content
         content = f"Filter by symbols: {self._symbols}" if self._symbols else ""
-        content += f"\n\n{self._generate_floating_shares([StockSymbol(symbol=s) for s in sorted(self._symbols)])}"
+        content_str, floating_shares = \
+            self._generate_floating_shares([StockSymbol(symbol=s) for s in sorted(self._symbols)])
+        content += f"\n\n{content_str}"
+        self._generate_floating_shares_summary_xlsx(floating_shares)
 
         # send email
-        self._email_api.send_email(self._alert_name, content, [self._file_8k_summary_filename])
+        self._email_api.send_email(self._alert_name, content,
+                                   [self._file_8k_summary_filename], [self._floating_shares_summary_filename])
 
         # send telegram message
         self._tg_bot.send_message(f"{self._alert_name}\n{content}")
-        self._tg_bot.send_file(self._file_8k_summary_filename, self._alert_name)
-        os.remove(self._file_8k_summary_filename)
+        for file in [self._file_8k_summary_filename, self._floating_shares_summary_filename]:
+            if os.path.exists(file):
+                self._tg_bot.send_file(file, self._alert_name)
+                os.remove(file)
 
 
 if __name__ == "__main__":
+    start = time.time()
     alert = FloatingSharesAlert("FloatingSharesAlert", symbols_file="floating_shares_symbols.txt")
     alert.run()
+
+    print(f"time taken: {time.time() - start} seconds")

smrti_quant_alerts/settings.py

@@ -6,8 +6,7 @@
 import pytz
 import datetime
 
-from binance.lib.utils import config_logging
-config_logging(logging, logging.WARNING)
+logging.basicConfig(level=logging.INFO)
 
 
 class Config:

smrti_quant_alerts/stock_crypto_api/stock_api.py

@@ -493,27 +493,38 @@ def get_all_8k_filings_for_today(self) -> List[Dict[str, Union[str, bool]]]:
                 symbol_set.remove(filing.get("symbol", None))
         return filtered_res
 
-    @error_handling("secapi", default_val=defaultdict(list))
+    @error_handling("financialmodelingprep", default_val=defaultdict(list))
     def get_floating_shares(self, stock_list: List[StockSymbol]) -> Dict[StockSymbol, List[Union[str, float]]]:
         """
         Get outstanding shares
 
         :param stock_list: [StockSymbol, ...]
-        :return: {StockSymbol: [period_new, shares_new, period_old, shares_old]}
+        :return: {StockSymbol: [date_new, shares_new, date_old, shares_old]}
         """
         res = defaultdict(list)
         for stock in stock_list:
-            api_url = f"{self.SEC_API_URL}/float?ticker={stock.ticker}&token={self.SEC_API_KEY}"
+            api_url = f"{self.FMP_API_URL}/v4/historical/shares_float?symbol={stock.ticker}&apikey={self.FMP_API_KEY}"
             response = requests.get(api_url, timeout=self.TIMEOUT).json()
-            data = response.get("data", [])
-            if len(data) >= 2:
-                for i in range(2):
-                    outstanding_shares = data[i].get("float", {}).get("outstandingShares", [])
-                    if len(outstanding_shares) == 0:
-                        res[stock] = []
+            if response:
+                try:
+                    date_now, floating_shares_now = \
+                        response[0].get("date", ""), int(response[0].get("floatShares", "0"))
+                    target_date_previous = \
+                        datetime.datetime.strptime(date_now, "%Y-%m-%d") - datetime.timedelta(days=90)
+                except ValueError:
+                    continue
+
+                for i in range(1, len(response)):
+                    try:
+                        date_previous_str, floating_shares_previous = response[i].get("date", ""), int(
+                            response[i].get("floatShares", "0"))
+                        date_previous = datetime.datetime.strptime(date_previous_str, "%Y-%m-%d")
+                    except ValueError:
+                        continue
+                    if date_previous <= target_date_previous:
+                        res[stock] = [date_now, floating_shares_now, date_previous_str, floating_shares_previous]
                         break
-                    res[stock].append(outstanding_shares[0].get("period", ""))
-                    res[stock].append(outstanding_shares[0].get("value", 0))
+
         return res
 
     @error_handling("financialmodelingprep", default_val=defaultdict(lambda: 0))