python websocket 获取火币 eth 历史行情

Posted 2021-06-18 tnan

import zlib
import websocket
import json
import redis
import pymysql
import time

rds = redis.StrictRedis(host=\'10.10.6.83\', port=6379, db=3, password="q123q123")
db = pymysql.connect("10.10.6.83", "root", "q123q123", "test")

# 使用cursor()方法创建一个游标对象
cursor = db.cursor()


# 火币历史行情数据获取
def huobi_history_market():
    # 火币有api可以获取行情数据，rest api 只提供了当前时间的行情，历史行情需要 websocket api
    url = \'wss://www.huobi.ge/-/s/pro/ws\'
    for i in range(5):
        try:
            # 创建websocket 连接
            ws = websocket.create_connection(url, timeout=5)
            break
        except:
            if i == 4:
                raise Exception("火币url可能被封了， 请求了5次都失败，换个url试试")
    while 1:
        # 获取上一次数据历史数据
        from_time = get_redis_last_time()
        to_time = from_time - 60 * 300
        if from_time <= int(time.mktime(time.strptime("2017-07-28 00:00:00", \'%Y-%m-%d %H:%M:%S\'))):
            exit(0)
        while 1:
            send_data = \'{"req":"market.ethusdt.kline.1min","symbol":"ethusdt","period":"1min","from":%d,"to":%d}\' % (
            to_time, from_time)

            # 发送请求获取数据
            ws.send(send_data)
            # 接收响应数据
            content_compress = ws.recv()
            # 数据解压缩
            content = zlib.decompress(content_compress, 16 + zlib.MAX_WBITS)

            # 数据转码
            content = content.decode()
            content = json.loads(content)
            # 判断数据是否正常获取到
            if content.get("status") == "ok":
                data = content.get("data")
                sql = " replace into eth_info(id, open, close,low,high,amount,vol,count) values "
                values_list = []
                for i in range(len(data)):
                    value = [str(data[i]["id"]), str(data[i]["open"]), str(data[i]["close"]), str(data[i]["low"]),
                             str(data[i]["high"]), str(data[i]["amount"]), str(data[i]["vol"]), str(data[i]["count"])]
                    values_list.append("(" + ", ".join(value) + ")")

                # 拼接sql
                sql += ", ".join(values_list)
                cursor.execute(sql)
                db.commit()
                rds.set("last_time", data[0]["id"])
                break


# 获取上一次爬取的时间
def get_redis_last_time():
    last_time = rds.get("last_time")
    if not last_time:
        last_time = get_mysql_last_time()
    if not last_time:
        last_time = get_now_time()
    if last_time:
        last_time = int(last_time)
    return last_time


def get_mysql_last_time():
    # 使用execute()方法执行SQL语句
    cursor.execute("SELECT id FROM eth_info order by id asc limit 0, 1")
    # 使用fetall()获取全部数据
    data = cursor.fetchall()
    if not data:
        return None
    return data[0][0]


def get_now_time():
    a = time.strftime(\'%Y-%m-%d %H:%M:00\', time.localtime(time.time() - 60))
    last_time = int(time.mktime(time.strptime(a, \'%Y-%m-%d %H:%M:%S\')))
    return last_time


if __name__ == \'__main__\':
    huobi_history_market()
    # 关闭游标和数据库的连接
    cursor.close()
    db.close()