最近BTC很火呀，下手玩了一些，但每天盯盤太累了，而大漲大跌的時(shí)候又往往會(huì)被錯(cuò)過

恰好昨天入了一個(gè)服務(wù)器，因此寫個(gè)程序放服務(wù)器上自動(dòng)監(jiān)測，那不就爽死了嗎

如果只是想下載代碼后運(yùn)行，可以直接去Github下載

文件目錄：

│  fangtang_push.py     #方糖推送服務(wù)接口的封裝
│  run.py               #主程序入口
└─util                  #工具包
    │  my_urllib.py     #服務(wù)器限制，重新封裝urllib
    │  time_foemat.py   #時(shí)間格式化

run.py：

程序流程：每10s獲取一次當(dāng)前15min和30min的線圖，計(jì)算本周期內(nèi)漲跌幅，如果到達(dá)漲跌上限，則調(diào)用方糖進(jìn)行推送，15min內(nèi)推送3次則休息至下個(gè)周期，30min內(nèi)推送5次也休息至下個(gè)周期。

代碼：

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

# 基礎(chǔ)API,需要科學(xué)上網(wǎng)
base_api = "https://api.huobi.pro/market/history"

# 15min漲跌提醒觸發(fā)值
line_15 = 0.55
# 30min漲跌提醒觸發(fā)值
line_30 = 0.75

#獲取當(dāng)前成交價(jià)格，以最近100單成交的均價(jià)為準(zhǔn)
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線圖數(shù)據(jù)
def get_kandle(period):
    url = base_api + "/kline?symbol=btcusdt&period=%dmin&size=1" % period
    data = myget(url)
    # 本周期開盤價(jià)
    open_price = int(data["data"][0]["open"])
    # 當(dāng)前價(jià)格
    close_price = int(data["data"][0]["close"])
    return open_price, close_price

# 處理獲取的K線圖數(shù)據(jù)，調(diào)用方糖進(jìn)行推送
def kandle_push(period, alarm_line, alarm_count, alarm_count_line):
    open_price, close_price = get_kandle(period)
    # 計(jì)算漲跌百分比
    current_line = (close_price - open_price) / open_price * 100
    # 超過漲跌上限，推送
    if abs(current_line) > alarm_line:
        current_price = get_current_price()
        detail = "%dmin 內(nèi)變動(dòng)了 %.2f%%\n\n當(dāng)前價(jià)格為%d\n當(dāng)前為此周期的第%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線圖數(shù)據(jù)
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

# 推送數(shù)量到達(dá)上限，停止推送
def sleep_until_next_period(period):
    period *= 60
    current_time = time.time()
    next_period = (int(current_time / period) + 1) * period
    title = "到達(dá) %dmin 內(nèi)推送上限，將休息至%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內(nèi)推送上限為3次，30min內(nèi)上限為5次
        # 到達(dá)上限則休息到下一個(gè)周期
        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獲取一次數(shù)據(jù)
        time.sleep(10)

其他說明：

一開始我是想有沒有個(gè)demo可以直接調(diào)用的，結(jié)果找了一圈找不到，大家寫的要么過期了，要么依賴的官方demo下線了（比如這個(gè)依賴的官方demo已經(jīng)下線了，而官方(Huobi-API)的另一個(gè)倉庫：https://github.com/huobiapi/Futures-Python-demo也已經(jīng)過期了），因此我找呀找，找到了官方的文檔，看了看文檔，又發(fā)現(xiàn)了一個(gè)官方的API測試平臺(tái)，依據(jù)這兩個(gè)，就可以愉快地?cái)]代碼了

官方文檔

API測試平臺(tái)

fangtang_push.py

調(diào)用方糖的服務(wù)，將數(shù)據(jù)推送給微信

需要先去方糖官網(wǎng)申請(qǐng)一個(gè)KEY，并綁定到自己的微信

代碼：

from util.my_urllib import mypost

# 自己去申請(qǐng)一個(gè)
API_Key=""

# 封裝方糖推送服務(wù)
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庫的，調(diào)用的時(shí)候直接requests.get()或者requests.post()簡簡單單，但是?。?！部署到服務(wù)器上后發(fā)現(xiàn)火幣API需要科學(xué)上網(wǎng)，于是我就部署了一下服務(wù)器的代理，結(jié)果部署好之后，Chrome是能訪問外網(wǎng)了，服務(wù)器上用到urllib的爬蟲全都掛掉了，提示信息：urlopen error EOF occurred in violation of protocol (_ssl.c:841)，然后再網(wǎng)上找到了大神的這個(gè)方法，爬蟲就復(fù)活了。

代碼：

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

# 重新搞一個(gè)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，返回的時(shí)候直接轉(zhuǎn)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

簡單地把時(shí)間戳轉(zhuǎn)成時(shí)間，便于查看

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