Monitoring Position Status
Polling for Quote Status
def poll_quote_status(token, temp_quote_id):
"""Poll for the status of a quote until it gets a permanent ID.""" #In order to track the status of the quote, we will poll the /instant_open/{address} endpoint
print(f"[STATUS] Starting to poll for quote status of temp ID: {temp_quote_id}")
headers = {
"Authorization": f"Bearer {token}"
}
max_attempts = 120 # 60 seconds (120 * 0.5s)
attempts = 0
if isinstance(temp_quote_id, str) and temp_quote_id.startswith('-'):
try:
temp_quote_id = int(temp_quote_id)
except ValueError:
print(f"[STATUS] Warning: Could not convert temp_quote_id {temp_quote_id} to int")
print(f"[STATUS] Looking for temp_quote_id: {temp_quote_id} (type: {type(temp_quote_id)})")
while attempts < max_attempts:
try:
response = requests.get(STATUS_URL, headers=headers)
if response.status_code != 200:
print(f"[STATUS] Error: Response status {response.status_code}")
print(f"[STATUS] Response text: {response.text}")
attempts += 1
time.sleep(CONFIG["STATUS_POLL_INTERVAL"])
continue
data = response.json()
print(f"[STATUS] Poll attempt {attempts+1}/{max_attempts}")
if not data:
print("[STATUS] Empty response, waiting for next update...")
attempts += 1
time.sleep(CONFIG["STATUS_POLL_INTERVAL"])
continue
quotes = []
if isinstance(data, list):
quotes = data
elif isinstance(data, dict) and "quotes" in data:
quotes = data["quotes"]
if not quotes:
print("[STATUS] No quotes found in response")
attempts += 1
time.sleep(CONFIG["STATUS_POLL_INTERVAL"])
continue
print(f"[STATUS] Found {len(quotes)} quotes to check")
# Look for any quote with a positive quote_id (confirmed)
for quote in quotes:
quote_id = quote.get("quote_id")
if isinstance(quote_id, str) and quote_id.isdigit():
quote_id = int(quote_id)
if isinstance(quote_id, int) and quote_id > 0:
print(f"[STATUS] ✓ CONFIRMED: Quote has permanent ID: {quote_id}")
return quote_id
attempts += 1
time.sleep(CONFIG["STATUS_POLL_INTERVAL"])
except Exception as e:
print(f"[ERROR] Error polling quote status: {e}")
traceback.print_exc()
attempts += 1
time.sleep(CONFIG["STATUS_POLL_INTERVAL"])
print("[STATUS] ⚠ Timed out waiting for permanent quote ID")
return NoneLast updated
