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:mel@1.94.96.154: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
心有猛虎,细嗅蔷薇