133 lines
4.3 KiB
Python
133 lines
4.3 KiB
Python
import time
|
|
import os
|
|
import asyncio
|
|
from datetime import datetime
|
|
from flextrade.flextrade_client import Client
|
|
from flextrade.constants.markets import (
|
|
BASE_MARKET_ETH_USD, BASE_MARKET_BTC_USD, BASE_MARKET_BNB_USD,
|
|
BASE_MARKET_SHIB_USD, BASE_MARKET_PEPE_USD, BASE_MARKET_SUI_USD,
|
|
BASE_MARKET_DOGE_USD, BASE_MARKET_AAVE_USD, BASE_MARKET_HBAR_USD,
|
|
BASE_MARKET_VIRTUAL_USD, BASE_MARKET_ADA_USD, BASE_MARKET_PENDLE_USD,
|
|
BASE_MARKET_TRX_USD, BASE_MARKET_AVAX_USD, BASE_MARKET_UNI_USD,
|
|
BASE_MARKET_SOL_USD, BASE_MARKET_LINK_USD, BASE_MARKET_XRP_USD,
|
|
BASE_MARKET_TON_USD
|
|
)
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
RPC_URL = os.getenv("RPC_URL")
|
|
PRIVATE_KEY = os.getenv("PRIVATE_KEY")
|
|
|
|
# List of all markets to test
|
|
MARKETS = [
|
|
BASE_MARKET_ETH_USD,
|
|
BASE_MARKET_BTC_USD,
|
|
BASE_MARKET_BNB_USD,
|
|
BASE_MARKET_SHIB_USD,
|
|
BASE_MARKET_PEPE_USD,
|
|
BASE_MARKET_SUI_USD,
|
|
BASE_MARKET_DOGE_USD,
|
|
BASE_MARKET_AAVE_USD,
|
|
BASE_MARKET_HBAR_USD,
|
|
BASE_MARKET_VIRTUAL_USD,
|
|
BASE_MARKET_ADA_USD,
|
|
BASE_MARKET_PENDLE_USD,
|
|
BASE_MARKET_TRX_USD,
|
|
BASE_MARKET_AVAX_USD,
|
|
BASE_MARKET_UNI_USD,
|
|
BASE_MARKET_SOL_USD,
|
|
BASE_MARKET_LINK_USD,
|
|
BASE_MARKET_XRP_USD,
|
|
BASE_MARKET_TON_USD
|
|
]
|
|
|
|
# Global variable to store latest prices with timestamps
|
|
latest_prices = {}
|
|
|
|
def print_market_info(market_info, debug=False):
|
|
if debug:
|
|
timestamp = datetime.now().strftime("%H:%M:%S")
|
|
print('[{0}] {1} : {2:.4f}'.format(timestamp, market_info["market"], market_info["price"]))
|
|
|
|
|
|
def market_info(debug=False):
|
|
"""Fetch market information once and update global prices"""
|
|
client = Client(
|
|
eth_private_key=PRIVATE_KEY,
|
|
rpc_url=RPC_URL
|
|
)
|
|
|
|
if debug:
|
|
print("Fetching market data...\n")
|
|
|
|
for market in MARKETS:
|
|
try:
|
|
market_info_data = client.public.get_market_info(market)
|
|
print_market_info(market_info_data, debug)
|
|
|
|
# Store the latest price and timestamp in global dictionary
|
|
market_name = market_info_data["market"]
|
|
current_timestamp = datetime.now()
|
|
latest_prices[market_name] = {
|
|
'price': market_info_data["price"],
|
|
'timestamp': current_timestamp,
|
|
'timestamp_str': current_timestamp.strftime("%Y-%m-%d %H:%M:%S")
|
|
}
|
|
|
|
time.sleep(2) # Wait 2 seconds between requests to avoid rate limiting
|
|
|
|
except Exception as e:
|
|
if debug:
|
|
print(f"Error fetching market {market}: {e}")
|
|
print('-' * 50)
|
|
# Skip this market and continue with the next one
|
|
continue
|
|
|
|
def market_info_loop(debug=False):
|
|
"""Continuously fetch market information in a loop"""
|
|
if debug:
|
|
print("Starting market data loop - press Ctrl+C to stop")
|
|
print("-" * 50)
|
|
|
|
try:
|
|
while True:
|
|
try:
|
|
market_info(debug)
|
|
# if debug:
|
|
# print(f"\nSleeping for 30 seconds...\n")
|
|
# time.sleep(30) # Wait 30 seconds between full cycles
|
|
except Exception as e:
|
|
if debug:
|
|
print(f"Error in market_info cycle: {e}")
|
|
print("Waiting 15 seconds before retry...")
|
|
time.sleep(15) # Wait 15 seconds before retrying on error
|
|
continue
|
|
# except KeyboardInterrupt:
|
|
# if debug:
|
|
# print("\nMarket data loop stopped by user")
|
|
except Exception as e:
|
|
if debug:
|
|
print(f"Fatal error occurred: {e}")
|
|
print("Market data loop stopped")
|
|
|
|
def get_latest_price(market_name):
|
|
"""Get the latest price for a specific market"""
|
|
market_data = latest_prices.get(market_name, None)
|
|
return market_data['price'] if market_data else None
|
|
|
|
def get_latest_price_with_timestamp(market_name):
|
|
"""Get the latest price with timestamp for a specific market"""
|
|
return latest_prices.get(market_name, None)
|
|
|
|
def get_all_latest_prices():
|
|
"""Get all latest prices with timestamps"""
|
|
return latest_prices.copy()
|
|
|
|
def get_all_latest_prices_only():
|
|
"""Get all latest prices without timestamps (backward compatibility)"""
|
|
return {market: data['price'] for market, data in latest_prices.items()}
|
|
|
|
if __name__ == '__main__':
|
|
# Enable debug when running directly
|
|
market_info_loop(debug=True) |