监测火币BTC价格，并推送到微信（2021.3.14）
2021-03-14 本文已影响0人
AndrewMyh
最近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
