diff --git a/data_fetcher.py b/data_fetcher.py index 7e6b7d8..e5c271e 100644 --- a/data_fetcher.py +++ b/data_fetcher.py @@ -18,7 +18,7 @@ from logging_utils import setup_logging class CandleFetcherDB: """ - Fetches 1-minute candle data and saves/updates it directly in an SQLite database. + Fetches 1-minute candle data and saves/updates it directly in a PostgreSQL database. """ def __init__(self, coins_to_fetch: list, interval: str, days_back: int): @@ -112,7 +112,7 @@ class CandleFetcherDB: df.sort_values(by='t', inplace=True) if not df.empty: - return self._save_to_sqlite_with_pandas(df, coin, table_existed) + return self._save_to_db_with_pandas(df, coin, table_existed) else: logging.info(f"No new candles to append for {coin}.") return 0 @@ -138,7 +138,8 @@ class CandleFetcherDB: max_retries = 3 for attempt in range(max_retries): try: - return self.info.candles_snapshot(coin, self.interval, start_ms, end_ms) + req = {"coin": coin, "interval": self.interval, "startTime": start_ms, "endTime": end_ms} + return self.info.post("/info", {"type": "candleSnapshot", "req": req}) except ClientError as e: if e.status_code == 429 and attempt < max_retries - 1: logging.warning("Rate limited. Retrying...") @@ -148,7 +149,7 @@ class CandleFetcherDB: return None return None - def _save_to_sqlite_with_pandas(self, df: pd.DataFrame, coin: str, is_append: bool) -> int: + def _save_to_db_with_pandas(self, df: pd.DataFrame, coin: str, is_append: bool) -> int: """Saves a pandas DataFrame to a PostgreSQL table and returns the number of saved rows.""" table_name = db.sanitize_table_name(coin, self.interval) try: @@ -175,7 +176,7 @@ class CandleFetcherDB: if __name__ == "__main__": - parser = argparse.ArgumentParser(description="Fetch historical candle data and save to SQLite.") + parser = argparse.ArgumentParser(description="Fetch historical candle data and save to PostgreSQL.") parser.add_argument( "--coins", nargs='+', diff --git a/data_fetcher_old.py b/data_fetcher_old.py index 5af87fd..90b3e73 100644 --- a/data_fetcher_old.py +++ b/data_fetcher_old.py @@ -162,7 +162,8 @@ class CandleFetcher: max_retries = 3 for attempt in range(max_retries): try: - return self.info.candles_snapshot(coin, self.interval, start_ms, end_ms) + req = {"coin": coin, "interval": self.interval, "startTime": start_ms, "endTime": end_ms} + return self.info.post("/info", {"type": "candleSnapshot", "req": req}) except ClientError as e: if e.status_code == 429 and attempt < max_retries - 1: logging.warning("Rate limited. Retrying in 2 seconds...") diff --git a/live_candle_fetcher.py b/live_candle_fetcher.py index 8cc1288..4a8d267 100644 --- a/live_candle_fetcher.py +++ b/live_candle_fetcher.py @@ -108,7 +108,8 @@ class LiveCandleFetcher: while current_start < end_ms: try: http_info = Info(constants.MAINNET_API_URL, skip_ws=True) - batch = http_info.candles_snapshot(coin, "1m", current_start, end_ms) + req = {"coin": coin, "interval": "1m", "startTime": current_start, "endTime": end_ms} + batch = http_info.post("/info", {"type": "candleSnapshot", "req": req}) if not batch: break @@ -158,11 +159,24 @@ class LiveCandleFetcher: print("\nListening for live candle data... Press Ctrl+C to stop.") try: while True: - time.sleep(1) + try: + time.sleep(1) + except Exception as e: + logging.error(f"WebSocket connection lost: {e}") + self.info.ws_manager.stop() + time.sleep(5) + self.info = Info(constants.MAINNET_API_URL, skip_ws=False) + for coin in self.coins_to_watch: + callback = lambda msg, c=coin: self.on_message({**msg, 'data': {**msg.get('data',{}), 'coin': c}}) + subscription = {"type": "candle", "coin": coin, "interval": "1m"} + self.info.ws_manager.subscribe(subscription, callback) + logging.info(f"Re-subscribed to 1m candles for {coin}") + time.sleep(0.2) + print("\nReconnected. Listening for live candle data...") except KeyboardInterrupt: print("\nStopping WebSocket listener...") self.info.ws_manager.stop() - self.candle_queue.put(None) + self.candle_queue.put(None) db_writer.join() print("Listener stopped.") diff --git a/scripts/gap_detector.py b/scripts/gap_detector.py index 68407fa..1cc17d4 100644 --- a/scripts/gap_detector.py +++ b/scripts/gap_detector.py @@ -87,7 +87,8 @@ def detect_and_fill_gaps(coin, conn): current_start = gap_start_ms while current_start < gap_end_ms: try: - batch = info.candles_snapshot(coin, "1m", current_start, gap_end_ms) + req = {"coin": coin, "interval": "1m", "startTime": current_start, "endTime": gap_end_ms} + batch = info.post("/info", {"type": "candleSnapshot", "req": req}) if not batch: break