Python实现http接口请求数据后，往RabbitMQ里面插入数据

python实现http接口请求数据服务后，往RABBITmq里面插入数据

import time
import requests
import pika
import datetime
base_url = 'https://www.okx.com'
api_url = '/api/v5/market/history-mark-price-candles'
# 时间颗粒度 [1m/3m/5m/15m/30m/1H/2H/4H]
time_unit = ['1m', '3m', '5m', '15m', '30m', '1h', '2h', '4h']
# 币种list
instId_list = ['ADA-USDT-SWAP', 'ANT-USDT-SWAP', 'ATOM-USDT-SWAP', 'BAT-USDT-SWAP', 'BCH-USDT-SWAP', 'BTC-USDT-SWAP', 'CFX-USDT-SWAP', 'DOGE-USDT-SWAP', 'ENS-USDT-SWAP', 'EOS-USDT-SWAP', 'ETC-USDT-SWAP', 'ETH-USDT-SWAP', 'GFT-USDT-SWAP', 'LTC-USDT-SWAP', 'ORDI-USDT-SWAP', 'STORJ-USDT-SWAP', 'TRX-USDT-SWAP', 'WLD-USDT-SWAP', 'XLM-USDT-SWAP', 'XRP-USDT-SWAP', 'YGG-USDT-SWAP']
# 循环时间颗粒度查询
for bar in time_unit:
    # 循环币种
    for line in instId_list:
        # 开始时间
        start_date = '2024-01-01 00:00:00'
        end_date = '2023-01-01 00:00:00'
        start_timestamp = datetime.datetime.strptime(start_date, '%Y-%m-%d %H:%M:%S').timestamp()
        end_timestamp = datetime.datetime.strptime(end_date, '%Y-%m-%d %H:%M:%S').timestamp()
        start_timestamp = int(start_timestamp * 1000)
        end_timestamp = int(end_timestamp * 1000)
        # 参数url
        instId = line
        after = start_timestamp
        bar = bar
        args_url = '?instId={}&after={}&bar={}'.format(instId, after, bar)
        url = base_url + api_url + args_url

        while after > end_timestamp:
            try:
                response = requests.get(url)
                result = response.json()
                if result['code'] == '0' and result['data']:
                    result.update({'instId': instId, 'bar': bar})
                    # 将result 写入到rabbitmq 实现逻辑
                    # RabbitMQ连接信息
                    serverURL = "amqp://mel:[email protected]:5672/okx_vhost"
                    exchangeName = "okx_all_exchange"
                    queueName = "okx_all_queue"
                    # 建立RabbitMQ连接
                    connection = pika.BlockingConnection(pika.URLParameters(serverURL))
                    channel = connection.channel()
                    # 声明exchange和queue
                    channel.exchange_declare(exchange=exchangeName, exchange_type='fanout', durable=True)
                    channel.queue_declare(queue=queueName, durable=True)
                    # 将数据写入RabbitMQ
                    channel.basic_publish(exchange=exchangeName, routing_key=queueName, body=str(result))
                    # 关闭连接
                    connection.close()
                    # 具体实现逻辑
                    after = result['data'][-1][0]
                    after = int(after)
                    str_date = datetime.datetime.fromtimestamp(after / 1000).strftime('%Y-%m-%d %H:%M:%S')
                    print('===bar===', bar)
                    print('===instId===', instId)
                    print('===str_date===', str_date)
                    # time.sleep(1)
                    # 更新参数url
                    args_url = '?instId={}&after={}&bar={}'.format(instId, after, bar)
                    url = base_url + api_url + args_url
                else:
                    break
            except requests.exceptions.ProxyError:
                print("ProxyError occurred. Retrying...")
                time.sleep(1)  # 延迟1秒后重试
                continue