Simon/MarketDataCollector
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

memory optimization

Browse Source
This commit is contained in:
Ajasra 2025-08-09 10:50:03 -05:00
parent 0c34623df8
commit d76d1a4925
2 changed files with 23 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
main.py
View File

@ -8,6 +8,10 @@ import time
import signal import signal
import queue import queue
# Memory tuning flags (set to True only if you need in-memory histories or latest snapshot)
ENABLE_HISTORY = False
STORE_LATEST_BOOK = False
latest_book = {'bids': [], 'asks': [], 'timestamp': None} latest_book = {'bids': [], 'asks': [], 'timestamp': None}
book_history = deque(maxlen=200) # Keep last 100 book updates (~20-60 seconds) book_history = deque(maxlen=200) # Keep last 100 book updates (~20-60 seconds)
trade_history = deque(maxlen=500) # Keep last 1000 trades (~2-10 minutes) trade_history = deque(maxlen=500) # Keep last 1000 trades (~2-10 minutes)
@ -134,13 +138,14 @@ def handle_message(data, instrument, db, trade_history, book_history, latest_boo
'side': trade.get('side'), 'side': trade.get('side'),
'timestamp': trade.get('ts') 'timestamp': trade.get('ts')
}) })
ts = float(trade.get('ts', time.time() * 1000)) if ENABLE_HISTORY:
trade_history.append({ ts = float(trade.get('ts', time.time() * 1000))
'price': trade.get('px'), trade_history.append({
'size': trade.get('sz'), 'price': trade.get('px'),
'side': trade.get('side'), 'size': trade.get('sz'),
'timestamp': ts 'side': trade.get('side'),
}) 'timestamp': ts
})
elif 'arg' in msg and msg['arg'].get('channel', '').startswith('books'): elif 'arg' in msg and msg['arg'].get('channel', '').startswith('books'):
for book in msg.get('data', []): for book in msg.get('data', []):
db.insert_book({ db.insert_book({
@ -149,15 +154,17 @@ def handle_message(data, instrument, db, trade_history, book_history, latest_boo
'asks': book.get('asks'), 'asks': book.get('asks'),
'timestamp': book.get('ts') 'timestamp': book.get('ts')
}) })
latest_book['bids'] = book.get('bids', []) if STORE_LATEST_BOOK:
latest_book['asks'] = book.get('asks', []) latest_book['bids'] = book.get('bids', [])
latest_book['timestamp'] = book.get('ts') latest_book['asks'] = book.get('asks', [])
ts = float(book.get('ts', time.time() * 1000)) latest_book['timestamp'] = book.get('ts')
book_history.append({ if ENABLE_HISTORY:
'bids': book.get('bids', []), ts = float(book.get('ts', time.time() * 1000))
'asks': book.get('asks', []), book_history.append({
'timestamp': ts 'bids': book.get('bids', []),
}) 'asks': book.get('asks', []),
'timestamp': ts
})
else: else:
logging.info(f"Unknown message for {instrument}: {msg}") logging.info(f"Unknown message for {instrument}: {msg}")
return True, now return True, now
@ -216,16 +223,6 @@ def main():
"BTC-USDT" "BTC-USDT"
] ]
# Configure logging to both file and stdout
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s',
handlers=[
logging.FileHandler('market_data_collector.log', mode='a'),
logging.StreamHandler()
]
)
try: try:
logging.info("Starting Market Data Collector") logging.info("Starting Market Data Collector")
logging.info(f"Initializing instruments: {instruments}") logging.info(f"Initializing instruments: {instruments}")

1
okx_client.py
View File

@ -4,7 +4,6 @@ import hmac
import hashlib import hashlib
import base64 import base64
import json import json
import pandas as pd
import threading import threading
import requests import requests
import websocket import websocket