Fix KeyError for prefixed coins in historical candle fetch
Bypass SDK's name_to_coin lookup in candles_snapshot by calling http_info.post directly with the raw coin name. This fixes KeyError for symbols like xyz:GOLD, xyz:CL, mkts:USTECH that aren't in the name_to_coin dictionary. The WebSocket subscription already bypassed this lookup (line 154), but the historical fetch path did not.
This commit is contained in:
@ -18,7 +18,7 @@ from logging_utils import setup_logging
|
|||||||
|
|
||||||
class CandleFetcherDB:
|
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):
|
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)
|
df.sort_values(by='t', inplace=True)
|
||||||
|
|
||||||
if not df.empty:
|
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:
|
else:
|
||||||
logging.info(f"No new candles to append for {coin}.")
|
logging.info(f"No new candles to append for {coin}.")
|
||||||
return 0
|
return 0
|
||||||
@ -138,7 +138,8 @@ class CandleFetcherDB:
|
|||||||
max_retries = 3
|
max_retries = 3
|
||||||
for attempt in range(max_retries):
|
for attempt in range(max_retries):
|
||||||
try:
|
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:
|
except ClientError as e:
|
||||||
if e.status_code == 429 and attempt < max_retries - 1:
|
if e.status_code == 429 and attempt < max_retries - 1:
|
||||||
logging.warning("Rate limited. Retrying...")
|
logging.warning("Rate limited. Retrying...")
|
||||||
@ -148,7 +149,7 @@ class CandleFetcherDB:
|
|||||||
return None
|
return None
|
||||||
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."""
|
"""Saves a pandas DataFrame to a PostgreSQL table and returns the number of saved rows."""
|
||||||
table_name = db.sanitize_table_name(coin, self.interval)
|
table_name = db.sanitize_table_name(coin, self.interval)
|
||||||
try:
|
try:
|
||||||
@ -175,7 +176,7 @@ class CandleFetcherDB:
|
|||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
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(
|
parser.add_argument(
|
||||||
"--coins",
|
"--coins",
|
||||||
nargs='+',
|
nargs='+',
|
||||||
|
|||||||
@ -162,7 +162,8 @@ class CandleFetcher:
|
|||||||
max_retries = 3
|
max_retries = 3
|
||||||
for attempt in range(max_retries):
|
for attempt in range(max_retries):
|
||||||
try:
|
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:
|
except ClientError as e:
|
||||||
if e.status_code == 429 and attempt < max_retries - 1:
|
if e.status_code == 429 and attempt < max_retries - 1:
|
||||||
logging.warning("Rate limited. Retrying in 2 seconds...")
|
logging.warning("Rate limited. Retrying in 2 seconds...")
|
||||||
|
|||||||
@ -108,7 +108,8 @@ class LiveCandleFetcher:
|
|||||||
while current_start < end_ms:
|
while current_start < end_ms:
|
||||||
try:
|
try:
|
||||||
http_info = Info(constants.MAINNET_API_URL, skip_ws=True)
|
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:
|
if not batch:
|
||||||
break
|
break
|
||||||
|
|
||||||
@ -158,7 +159,20 @@ class LiveCandleFetcher:
|
|||||||
print("\nListening for live candle data... Press Ctrl+C to stop.")
|
print("\nListening for live candle data... Press Ctrl+C to stop.")
|
||||||
try:
|
try:
|
||||||
while True:
|
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:
|
except KeyboardInterrupt:
|
||||||
print("\nStopping WebSocket listener...")
|
print("\nStopping WebSocket listener...")
|
||||||
self.info.ws_manager.stop()
|
self.info.ws_manager.stop()
|
||||||
|
|||||||
@ -87,7 +87,8 @@ def detect_and_fill_gaps(coin, conn):
|
|||||||
current_start = gap_start_ms
|
current_start = gap_start_ms
|
||||||
while current_start < gap_end_ms:
|
while current_start < gap_end_ms:
|
||||||
try:
|
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:
|
if not batch:
|
||||||
break
|
break
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user