transfer to aws

ws_rtds.py

@@ -12,7 +12,7 @@ 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
 
 ### Allow only ipv4 ###
 def allowed_gai_family():
@@ -21,7 +21,11 @@ urllib3_cn.allowed_gai_family = allowed_gai_family
 
 ### Database ###
 USE_DB: bool = True
+USE_VK: bool = True
+VK_CHANNEL = 'poly_rtds_cl_btcusd'
 CON: AsyncContextManager | None = None
+VAL_KEY = None
+
 
 ### Logging ###
 LOG_FILEPATH: str = '/root/logs/Polymarket_RTDS.log'
@@ -39,11 +43,13 @@ async def create_rtds_btcusd_table(
         logging.info("NO DB CONNECTION, SKIPPING Create Statements")
     else:
         if engine == 'mysql':
+            logging.info('Creating Table if Does Not Exist: poly_rtds_cl_btcusd')
             await CON.execute(text("""
-            CREATE TABLE IF NOT EXISTS poly_rtds_btcusd_cl (
+            CREATE TABLE IF NOT EXISTS poly_rtds_cl_btcusd (
+                timestamp_arrival BIGINT,
                 timestamp_msg BIGINT,
                 timestamp_value BIGINT,
-                value DOUBLE,
+                value DOUBLE
             );
             """))
             await CON.commit()
@@ -51,6 +57,7 @@ async def create_rtds_btcusd_table(
             raise ValueError('Only MySQL engine is implemented')
 
 async def insert_rtds_btcusd_table(
+        timestamp_arrival: int,
         timestamp_msg: int,
         timestamp_value: int,
         value: int,
@@ -58,6 +65,7 @@ async def insert_rtds_btcusd_table(
         engine: str = 'mysql', # mysql | duckdb
     ) -> None:
     params={
+        'timestamp_arrival': timestamp_arrival,
         'timestamp_msg': timestamp_msg,
         'timestamp_value': timestamp_value,
         'value': value,
@@ -67,14 +75,16 @@ async def insert_rtds_btcusd_table(
     else:
         if engine == 'mysql':
             await CON.execute(text("""
-            INSERT INTO poly_rtds_btcusd_cl 
+            INSERT INTO poly_rtds_cl_btcusd 
             (
+                timestamp_arrival,
                 timestamp_msg,
                 timestamp_value,
                 value
             )
             VALUES 
                 (
+                :timestamp_arrival,
                 :timestamp_msg,
                 :timestamp_value,
                 :value
@@ -91,7 +101,7 @@ async def insert_rtds_btcusd_table(
 async def rtds_stream():
     global HIST_TRADES
 
-    async with websockets.connect(WSS_URL) as websocket:
+    async for websocket in websockets.connect(WSS_URL):
         logging.info(f"Connected to {WSS_URL}")
 
         subscribe_msg = {
@@ -113,9 +123,26 @@ async def rtds_stream():
                     try: 
                         data = json.loads(message)
                         if data['payload'].get('value', None) is not None:
-                            print(f'🤑 BTC Chainlink Last Px: {data['payload']['value']:_.4f}; TS: {pd.to_datetime(data['timestamp'], unit='ms')}')
+                            ts_arrival = round(datetime.now().timestamp()*1000)
+                            print(f'🤑 BTC Chainlink Last Px: {data['payload']['value']:_.4f}; TS: {pd.to_datetime(data['payload']['timestamp'], unit='ms')}')
+                            VAL_KEY_OBJ = json.dumps({
+                                'timestamp_arrival': ts_arrival,
+                                'timestamp_msg': data['timestamp'],
+                                'timestamp_value': data['payload']['timestamp'],
+                                'value': data['payload']['value'],
+                            })
+                            VAL_KEY.publish(VK_CHANNEL, VAL_KEY_OBJ)
+                            VAL_KEY.set(VK_CHANNEL, VAL_KEY_OBJ)
+                            await insert_rtds_btcusd_table(
+                                CON=CON,
+                                timestamp_arrival=ts_arrival,
+                                timestamp_msg=data['timestamp'],
+                                timestamp_value=data['payload']['timestamp'],
+                                value=data['payload']['value'],
+                            )
                         else:
-                            logging.info(f'Initial or unexpected data struct, skipping: {data}')
+                            # logging.info(f'Initial or unexpected data struct, skipping: {data}')
+                            logging.info('Initial or unexpected data struct, skipping')
                             continue
                     except (json.JSONDecodeError, ValueError):
                         logging.warning(f'Message not in JSON format, skipping: {message}')
@@ -125,17 +152,28 @@ async def rtds_stream():
         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())
 
 
 async def main():
+    global VAL_KEY
     global CON
+    
+    if USE_VK:
+        VAL_KEY = valkey.Valkey(host='localhost', port=6379, db=0)
+        published_count = VAL_KEY.publish(VK_CHANNEL,f"Hola, starting to publish to valkey: {VK_CHANNEL} @ {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+        logging.info(f"Valkey message published to {published_count} subscribers of {VK_CHANNEL}")
+    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/mkt_maker')
+        engine = create_async_engine('mysql+asyncmy://root:pwd@localhost/polymarket')
         async with engine.connect() as CON:
+            await create_rtds_btcusd_table(CON=CON)
             await rtds_stream()
     else:
         CON = None