60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to check live updates API
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
import time
|
|
from datetime import datetime
|
|
|
|
def test_live_updates():
|
|
"""Test the live-updates-batch API"""
|
|
|
|
print("Testing live-updates-batch API...")
|
|
print("-" * 50)
|
|
|
|
url = "http://localhost:8051/api/live-updates-batch"
|
|
|
|
payload = {
|
|
"symbol": "ETH/USDT",
|
|
"timeframes": ["1s", "1m"]
|
|
}
|
|
|
|
# Make 3 requests with 3 second intervals to see if data changes
|
|
for i in range(3):
|
|
try:
|
|
print(f"\n=== Request {i+1} at {datetime.now().strftime('%H:%M:%S')} ===")
|
|
|
|
response = requests.post(url, json=payload, timeout=10)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
|
|
print(f"Success: {data.get('success')}")
|
|
print(f"Server time: {data.get('server_time')}")
|
|
|
|
if data.get('success') and data.get('chart_updates'):
|
|
for timeframe, update in data['chart_updates'].items():
|
|
candle = update.get('candle', {})
|
|
timestamp = candle.get('timestamp', 'N/A')
|
|
close_price = candle.get('close', 'N/A')
|
|
is_confirmed = update.get('is_confirmed', False)
|
|
|
|
print(f" {timeframe}: {timestamp} | Close: {close_price} | Confirmed: {is_confirmed}")
|
|
else:
|
|
print(f" Error: {data.get('error', 'Unknown error')}")
|
|
else:
|
|
print(f" HTTP Error: {response.status_code}")
|
|
print(f" Response: {response.text}")
|
|
|
|
except Exception as e:
|
|
print(f" Request failed: {e}")
|
|
|
|
if i < 2: # Don't sleep after last request
|
|
time.sleep(3)
|
|
|
|
print("\nTest completed!")
|
|
|
|
if __name__ == "__main__":
|
|
test_live_updates() |