监测火币BTC价格，并推送到微信（2021.3.14）

最近BTC很火呀，下手玩了一些，但每天盯盘太累了，而大涨大跌的时候又往往会被错过

恰好昨天入了一个服务器，因此写个程序放服务器上自动监测，那不就爽死了吗

如果只是想下载代码后运行，可以直接去Github下载

文件目录：

│  fangtang_push.py     #方糖推送服务接口的封装
│  run.py               #主程序入口
└─util                  #工具包
    │  my_urllib.py     #服务器限制，重新封装urllib
    │  time_foemat.py   #时间格式化

run.py：

程序流程：每10s获取一次当前15min和30min的线图，计算本周期内涨跌幅，如果到达涨跌上限，则调用方糖进行推送，15min内推送3次则休息至下个周期，30min内推送5次也休息至下个周期。

代码：

import time
import requests
from fangtang_push import fangtang_push
from util.time_format import timestamp_to_date
from util.my_urllib import myget

# 基础API,需要科学上网
base_api = "https://api.huobi.pro/market/history"

# 15min涨跌提醒触发值
line_15 = 0.55
# 30min涨跌提醒触发值
line_30 = 0.75

#获取当前成交价格，以最近100单成交的均价为准
def get_current_price():
    url = base_api + "/trade?symbol=btcusdt&size=100"
    data = myget(url)
    data = data["data"]
    average_price = 0
    for i in data:
        average_price += i["data"][0]["price"]
    average_price /= 100
    average_price = int(average_price)
    return average_price

# 获取K线图数据
def get_kandle(period):
    url = base_api + "/kline?symbol=btcusdt&period=%dmin&size=1" % period
    data = myget(url)
    # 本周期开盘价
    open_price = int(data["data"][0]["open"])
    # 当前价格
    close_price = int(data["data"][0]["close"])
    return open_price, close_price

# 处理获取的K线图数据，调用方糖进行推送
def kandle_push(period, alarm_line, alarm_count, alarm_count_line):
    open_price, close_price = get_kandle(period)
    # 计算涨跌百分比
    current_line = (close_price - open_price) / open_price * 100
    # 超过涨跌上限，推送
    if abs(current_line) > alarm_line:
        current_price = get_current_price()
        detail = "%dmin 内变动了 %.2f%%\n\n当前价格为%d\n当前为此周期的第%d/%d次推送" \
                 % (period, current_line, current_price, alarm_count, alarm_count_line)
        if current_line > 0:
            fangtang_push("涨涨涨涨涨涨涨涨涨！", detail)
        else:
            fangtang_push("跌跌跌跌跌跌跌跌跌！", detail)
        return True
    else:
        return False

# 分别处理15min和30min的k线图数据
def call_kandle_push(alarm_15_count, alarm_30_count):
    if kandle_push(15, line_15, alarm_15_count, alarm_count_line=3):
        return True, True
    else:
        if kandle_push(30, line_30, alarm_30_count, alarm_count_line=5):
            return False, True
        else:
            return False, False

# 推送数量到达上限，停止推送
def sleep_until_next_period(period):
    period *= 60
    current_time = time.time()
    next_period = (int(current_time / period) + 1) * period
    title = "到达 %dmin 内推送上限，将休息至%s" % (period / 60, timestamp_to_date(next_period))
    print(title)
    fangtang_push(title, "")
    time_gap = next_period - current_time
    time.sleep(time_gap)

if __name__ == '__main__':
    alarm_15_count = 1
    alarm_30_count = 1
    while True:
        # 15min内推送上限为3次，30min内上限为5次
        # 到达上限则休息到下一个周期
        alarm_15, alarm_30 = call_kandle_push(alarm_15_count, alarm_30_count)
        if alarm_15_count == 3:
            alarm_15_count = 1
            sleep_until_next_period(15)
            continue
        if alarm_30_count == 5:
            alarm_30_count = 1
            sleep_until_next_period(30)
            continue
        if alarm_15:
            alarm_15_count += 1
        if alarm_30:
            alarm_30_count += 1
        # 每10s获取一次数据
        time.sleep(10)

其他说明：

一开始我是想有没有个demo可以直接调用的，结果找了一圈找不到，大家写的要么过期了，要么依赖的官方demo下线了（比如这个依赖的官方demo已经下线了，而官方(Huobi-API)的另一个仓库：https://github.com/huobiapi/Futures-Python-demo也已经过期了），因此我找呀找，找到了官方的文档，看了看文档，又发现了一个官方的API测试平台，依据这两个，就可以愉快地撸代码了

官方文档 API测试平台

fangtang_push.py

调用方糖的服务，将数据推送给微信

需要先去方糖官网申请一个KEY，并绑定到自己的微信

代码：

from util.my_urllib import mypost

# 自己去申请一个
API_Key=""

# 封装方糖推送服务
def fangtang_push(title, detail):
    push_data = {
        "title": title,
        "desp": detail
    }
    headers = {
        'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_4) AppleWebKit/537.36 (KHTML, like Gecko) '
                      'Chrome/52.0.2743.116 Safari/537.36',
        'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'
    }
    push_url = "https://sctapi.ftqq.com/%s.send"%API_Key
    mypost(push_url, push_data=push_data,headers=headers)

if __name__ == '__main__':
    fangtang_push("测试推送","哈哈哈")
    pass

my_urllib.py

原本我是直接使用requests库的，调用的时候直接requests.get()或者requests.post()简简单单，但是！！！部署到服务器上后发现火币API需要科学上网，于是我就部署了一下服务器的代理，结果部署好之后，Chrome是能访问外网了，服务器上用到urllib的爬虫全都挂掉了，提示信息：urlopen error EOF occurred in violation of protocol (_ssl.c:841)，然后再网上找到了大神的这个方法，爬虫就复活了。

代码：

import urllib.request
from bs4 import BeautifulSoup
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.poolmanager import PoolManager
import ssl

# 重新搞一个Adapter
class MyAdapter(HTTPAdapter):
    def init_poolmanager(self, connections, maxsize, block=False):
        self.poolmanager = PoolManager(num_pools=connections,
                                       maxsize=maxsize,
                                       block=block,
                                       ssl_version=ssl.PROTOCOL_TLSv1)
      
s = requests.Session()
s.mount('https://', MyAdapter())
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:23.0) Gecko/20100101 Firefox/23.0'}

# 自己搞get，返回的时候直接转dict
def myget(url, headers=headers):
    req = urllib.request.Request(url=url, headers=headers)
    response = urllib.request.urlopen(req)
    return eval(response.read())

# 自己搞post
def mypost(url: str, push_data: dict, headers: dict = headers) -> dict:
    data = bytes(urllib.parse.urlencode(push_data), encoding='utf8')
    req = urllib.request.Request(url=url, headers=headers)
    response = urllib.request.urlopen(req, data=data)
    return eval(response.read())

if __name__ == '__main__':
    url = "https://api.huobi.pro/market/history/kline?symbol=btcusdt&period=30min&size=1"
    r = myget(url)
    print(r)

time_format.py

简单地把时间戳转成时间，便于查看

import time

def timestamp_to_date(time_stamp, format_string="%H:%M"):
    time_array = time.localtime(time_stamp)
    str_date = time.strftime(format_string, time_array)
    return str_date

if __name__ == '__main__':
    pass