监测火币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
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 220,458评论 6 513
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 94,030评论 3 396
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 166,879评论 0 358
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 59,278评论 1 295
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 68,296评论 6 397
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 52,019评论 1 308
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 40,633评论 3 420
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 39,541评论 0 276
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 46,068评论 1 319
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 38,181评论 3 340
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 40,318评论 1 352
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 35,991评论 5 347
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 41,670评论 3 331
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 32,183评论 0 23
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 33,302评论 1 272
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 48,655评论 3 375
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 45,327评论 2 358

推荐阅读更多精彩内容