refactor algo orchestrator and hedge ratio bug fix

ws_aster_fund_rate_all.py

@@ -0,0 +1,110 @@
+import asyncio
+import json
+import logging
+import socket
+import traceback
+from datetime import datetime, timezone
+from typing import AsyncContextManager
+import math
+
+import numpy as np
+import pandas as pd
+import requests.packages.urllib3.util.connection as urllib3_cn  # type: ignore
+from sqlalchemy import text
+import websockets
+from sqlalchemy.ext.asyncio import create_async_engine
+import valkey
+import os
+from dotenv import load_dotenv
+import modules.utils as utils
+
+### Allow only ipv4 ###
+def allowed_gai_family():
+    return socket.AF_INET
+urllib3_cn.allowed_gai_family = allowed_gai_family
+
+### Database ###
+USE_DB: bool = False
+USE_VK: bool = True
+VK_FUND_RATE_ALL = 'fund_rate_aster_all'
+
+CON: AsyncContextManager | None = None
+VAL_KEY = None
+
+### Logging ###
+load_dotenv()
+LOG_FILEPATH: str = os.getenv("LOGS_PATH") + '/Fund_Rate_Aster_FR_ALL.log'
+
+### Globals ###
+WSS_URL = "wss://fstream.asterdex.com/ws/!markPrice@arr"
+
+
+### Websocket ###
+async def ws_stream():
+    async for websocket in websockets.connect(WSS_URL):
+        logging.info(f"Connected to {WSS_URL}")
+        try:
+            async for message in websocket:
+                if isinstance(message, str):
+                    try: 
+                        data = json.loads(message)
+                        if data:
+                            VAL_KEY.set(VK_FUND_RATE_ALL, json.dumps(data))
+                            # print(f'VK_SAVED: {len(data)}')
+                            continue
+                        else:
+                            logging.info(f'Initial or unexpected data struct, skipping: {data}')
+                            continue
+                    except (json.JSONDecodeError, ValueError):
+                        logging.warning(f'Message not in JSON format, skipping: {message}')
+                        continue
+                else:
+                    raise ValueError(f'Type: {type(data)}  not expected: {message}')
+        except websockets.ConnectionClosed as e:
+            logging.error(f'Connection closed: {e}')
+            logging.error(traceback.format_exc())
+            continue
+        except Exception as e:
+            logging.error(f'Connection closed: {e}')
+            logging.error(traceback.format_exc())
+
+### Startup ###
+async def main():
+    global VAL_KEY
+    global CON
+    
+    if USE_VK:
+        VAL_KEY = valkey.Valkey(host='localhost', port=6379, db=0)
+    else:
+        VAL_KEY = None
+        logging.warning("VALKEY NOT BEING USED, NO DATA WILL BE PUBLISHED")
+
+    if USE_DB:
+        engine = create_async_engine('mysql+asyncmy://root:pwd@localhost/fund_rate')
+        async with engine.connect() as CON:
+            # await create_rtds_btcusd_table(CON=CON)
+            await ws_stream()
+    else:
+        CON = None
+        logging.warning("DATABASE NOT BEING USED, NO DATA WILL BE RECORDED")
+        await ws_stream()
+
+
+if __name__ == '__main__':
+    START_TIME = round(datetime.now().timestamp()*1000)
+
+    logging.info(f'Log FilePath: {LOG_FILEPATH}')
+
+    logging.basicConfig(
+        force=True,
+        filename=LOG_FILEPATH,
+        level=logging.INFO,
+        format='%(asctime)s - %(levelname)s - %(message)s',
+        filemode='w'
+    )
+    logging.info(f"STARTED: {START_TIME}")
+    
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        logging.info("Stream stopped")