5bb78bd9be
- Create project structure with FastAPI - Add database models for blocks, transactions, arbitrages, pools, and DEXes - Implement Solana RPC client for fetching blockchain data - Create arbitrage detection algorithm - Implement comprehensive API endpoints for analytics - Set up database migrations with Alembic - Add detailed project documentation generated with BackendIM... (backend.im) Co-Authored-By: Claude <noreply@anthropic.com>
120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
import threading
|
|
import time
|
|
from typing import Optional
|
|
|
|
from loguru import logger
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.session import SessionLocal
|
|
from app.services.blockchain_fetcher import BlockchainFetcher
|
|
from app.services.arbitrage_detector import ArbitrageDetector
|
|
from app.core.config import settings
|
|
|
|
|
|
class BackgroundWorker:
|
|
"""Background worker for fetching blockchain data and analyzing arbitrages."""
|
|
|
|
def __init__(self):
|
|
"""Initialize the background worker."""
|
|
self._fetcher_thread = None
|
|
self._detector_thread = None
|
|
self._stop_event = threading.Event()
|
|
|
|
def start_fetcher(self):
|
|
"""Start the blockchain fetcher in a background thread."""
|
|
if self._fetcher_thread and self._fetcher_thread.is_alive():
|
|
logger.warning("Blockchain fetcher already running")
|
|
return False
|
|
|
|
self._stop_event.clear()
|
|
self._fetcher_thread = threading.Thread(target=self._run_fetcher)
|
|
self._fetcher_thread.daemon = True
|
|
self._fetcher_thread.start()
|
|
|
|
logger.info("Blockchain fetcher started")
|
|
return True
|
|
|
|
def start_detector(self):
|
|
"""Start the arbitrage detector in a background thread."""
|
|
if self._detector_thread and self._detector_thread.is_alive():
|
|
logger.warning("Arbitrage detector already running")
|
|
return False
|
|
|
|
self._stop_event.clear()
|
|
self._detector_thread = threading.Thread(target=self._run_detector)
|
|
self._detector_thread.daemon = True
|
|
self._detector_thread.start()
|
|
|
|
logger.info("Arbitrage detector started")
|
|
return True
|
|
|
|
def stop(self):
|
|
"""Stop all background threads."""
|
|
if not (self._fetcher_thread or self._detector_thread):
|
|
logger.warning("No background threads are running")
|
|
return False
|
|
|
|
self._stop_event.set()
|
|
|
|
if self._fetcher_thread:
|
|
self._fetcher_thread.join(timeout=5.0)
|
|
self._fetcher_thread = None
|
|
|
|
if self._detector_thread:
|
|
self._detector_thread.join(timeout=5.0)
|
|
self._detector_thread = None
|
|
|
|
logger.info("All background threads stopped")
|
|
return True
|
|
|
|
def _run_fetcher(self):
|
|
"""Run the blockchain fetcher continuously."""
|
|
db = SessionLocal()
|
|
try:
|
|
fetcher = BlockchainFetcher(db)
|
|
|
|
while not self._stop_event.is_set():
|
|
try:
|
|
# Fetch latest blocks
|
|
blocks = fetcher.fetch_latest_blocks()
|
|
logger.info(f"Fetched {len(blocks)} new blocks")
|
|
|
|
# Sleep for the configured interval
|
|
self._stop_event.wait(settings.POLLING_INTERVAL)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in blockchain fetcher: {str(e)}")
|
|
# Sleep a bit before retrying
|
|
self._stop_event.wait(10)
|
|
|
|
finally:
|
|
db.close()
|
|
logger.info("Blockchain fetcher thread stopped")
|
|
|
|
def _run_detector(self):
|
|
"""Run the arbitrage detector continuously."""
|
|
db = SessionLocal()
|
|
try:
|
|
detector = ArbitrageDetector(db)
|
|
|
|
while not self._stop_event.is_set():
|
|
try:
|
|
# Process blocks that haven't been analyzed yet
|
|
unprocessed_count = detector.detect_arbitrages()
|
|
logger.info(f"Analyzed {unprocessed_count} blocks for arbitrages")
|
|
|
|
# Sleep a bit to avoid hammering the database
|
|
self._stop_event.wait(10)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in arbitrage detector: {str(e)}")
|
|
# Sleep a bit before retrying
|
|
self._stop_event.wait(10)
|
|
|
|
finally:
|
|
db.close()
|
|
logger.info("Arbitrage detector thread stopped")
|
|
|
|
|
|
# Singleton instance for the application
|
|
worker = BackgroundWorker() |