Grenade_LOGO_橫1
Instagram Facebook Telegram

Grenade_LOGO_橫1

Grenade_LOGO_橫1

發佈於 作者: Lee Chun Hao

如何利用 Python + IFTTT 製作個人專屬 LINE 行情提醒機器人？

目前數字貨幣市場中，提醒工具、APP 可謂是琳瑯滿目，幾乎各大數據網站、錢包乃至於交易所都有提供相應的行情提醒服務，然而這些服務商除了 TradingView 付費版以外，大多數功能僅限於價格點位、行情劇烈波動等，對於技術指標等等數據是相對缺乏的，難以滿足交易者的需求。

對於剛進入市場還沒開始賺錢的人，購買 TradingView 的性價比又難以衡量。作為一個「客家人」，遊戲還沒開始玩就要先氪金肯定是不能接受的！所以這篇文將會利用 Python 爬蟲透過 IFTTT 與 LINE Notify，在不用花錢的情況下，實現「觸發某行情事件，就自動發送 LINE 訊息」的功能。

IFTTT

IFTTT 是個網路服務串接平台，名字為「 IF THIS, THEN THAT」 的縮寫，代表「若發生這個事件 ( THIS )，則 ( THEN ) 執行那個動作 (THAT) 」，在 IFTTT 中大多數服務也都是用這種模式在進行，你可以透過 IFTTT，用很簡單的方式串接許多知名平台的服務，例如 Gmail、Twitter、Facebook…等，當然也包括了本篇會用到的 WebHook + Line。

IFTTT 網站：https://ifttt.com/

進入 IFTTT 網站後請點選右上角「Sign up」註冊帳號並登入，如果懶得註冊帳號 IFTTT 也有支援 Apple、Google 、 Facebook 等第三方登入，登入後上方選單有「Create」點選其中的「Applets」來創建我們需要串聯的服務內容，THIS 位請搜尋並選擇 WebHook、THAT 位請搜尋並選擇  LINE。

請點選首頁右上角「Sign up」註冊並登入
點選「Create > Applets」來創建服務
在 THIS 位選擇「WebHook」並為事件命名
在 THAT 位選擇「LINE」並設定接受信息的群組或聊天以及訊息格式

個人習慣的訊息格式如下：

通知事件名稱<br><br>
{{Value1}}

在 Step 6 頁面按下「 Finish」之後，IFTTT的設定就先告一段落了。

Python + WebHook

首先，請自行準備 Python 環境以及編譯器 ( 網路有許多相關教學 ) ，並在終端機輸入以下指令安裝行情計算相關套件：

pip install numpy
pip install pandas
pip install TA-Lib

在設定好環境之後，我們將透過以下步驟一步一步的完成行情數據計算並呼叫 Line Notify：

    • 獲取 K 線數據
      • 選定交易所 ( API URL )
      • 選定交易對 ( BTC/USDT )
      • 透過爬蟲將資料儲存至 Pandas DataFrame
    • 透過 TA-LIB 套件運算行情數據
      • RSI
      • MA
    • 傳送到 Line Notify
      • 行情邏輯判斷
      • 傳送至 WebHook

獲取 K 線數據 ( 以幣安為例 )

import pandas as pd
import requests
from datetime import datetime

# # DataFrame Setting
pd.set_option('display.max_columns', None)
pd.set_option('display.max_rows', None)
pd.set_option('max_colwidth',100)
pd.set_option('display.width', 5000)
pd.set_option('display.unicode.ambiguous_as_wide', True)
pd.set_option('display.unicode.east_asian_width', True)

# # Global Variables Setting
url  = 'https://api.binance.com/'
coin = 'BTCUSDT'

# # Get Market Data
def GetKline(url, symbol, interval):
    try:
        data = requests.get(url + 'api/v3/klines', params={'symbol': symbol, 'interval': interval, 'limit': 1000}).json()
    except Exception as e:
        print ('Error! problem is {}'.format(e.args[0]))
    tmp  = []
    pair = []
    for base in data:
        tmp  = []
        for i in range(0,6):
            if i == 0:
                base[i] = datetime.fromtimestamp(base[i]/1000)
            tmp.append(base[i])
        pair.append(tmp)
    df = pd.DataFrame(pair, columns=['date', 'open', 'high', 'low', 'close', 'volume'])
    df.date = pd.to_datetime(df.date)
    df.set_index("date", inplace=True)
    df = df.astype(float)
    return df

if __name__ == "__main__":
    kline = GetKline(url, coin, '1d')
    print (kline)
打印出來的 K 線數據

透過 TA-LIB 套件運算行情數據

# # RSI 計算

import pandas as pd
import requests

from talib    import abstract
from datetime import datetime

# # DataFrame Setting
pd.set_option('display.max_columns', None)
pd.set_option('display.max_rows', None)
pd.set_option('max_colwidth',100)
pd.set_option('display.width', 5000)
pd.set_option('display.unicode.ambiguous_as_wide', True)
pd.set_option('display.unicode.east_asian_width', True)

# # Global Variables Setting
url  = 'https://api.binance.com/'
coin = 'BTCUSDT'

# # Line Notify
def send_ifttt(msg):   
    url = ('https://maker.ifttt.com/trigger/事件名稱/with/'+'key/金鑰'+'?value1='+str(msg))
    r = requests.get(url)      
    if r.text[:5] == 'Congr':  
        print('Success send (' +str(msg)+') to Line')
    return r.text

# # Get Market Data
def GetKline(url, symbol, interval):
    try:
        data = requests.get(url + 'api/v3/klines', params={'symbol': symbol, 'interval': interval, 'limit': 1000}).json()
    except Exception as e:
        print ('Error! problem is {}'.format(e.args[0]))
    tmp  = []
    pair = []
    for base in data:
        tmp  = []
        for i in range(0,6):
            if i == 0:
                base[i] = datetime.fromtimestamp(base[i]/1000)
            tmp.append(base[i])
        pair.append(tmp)
    df = pd.DataFrame(pair, columns=['date', 'open', 'high', 'low', 'close', 'volume'])
    df.date = pd.to_datetime(df.date)
    df.set_index("date", inplace=True)
    df = df.astype(float)
    return df

# # Financial indicators
def RSI(df, period):
    return abstract.RSI(df, timeperiod=period)

if __name__ == "__main__":
    kline = GetKline(url, coin, '1d')
    index = RSI(kline, 14)
    print (index)
利用幣安日線數據計算 RSI
# # 計算 200 MA

import pandas as pd
import requests

from talib    import abstract
from datetime import datetime

# # DataFrame Setting
pd.set_option('display.max_columns', None)
pd.set_option('display.max_rows', None)
pd.set_option('max_colwidth',100)
pd.set_option('display.width', 5000)
pd.set_option('display.unicode.ambiguous_as_wide', True)
pd.set_option('display.unicode.east_asian_width', True)

# # Global Variables Setting
url  = 'https://api.binance.com/'
coin = 'BTCUSDT'

# # Line Notify
def send_ifttt(msg):   
    url = ('https://maker.ifttt.com/trigger/事件名稱/with/'+'key/金鑰'+'?value1='+str(msg))
    r = requests.get(url)      
    if r.text[:5] == 'Congr':  
        print('Success send (' +str(msg)+') to Line')
    return r.text

# # Get Market Data
def GetKline(url, symbol, interval):
    try:
        data = requests.get(url + 'api/v3/klines', params={'symbol': symbol, 'interval': interval, 'limit': 1000}).json()
    except Exception as e:
        print ('Error! problem is {}'.format(e.args[0]))
    tmp  = []
    pair = []
    for base in data:
        tmp  = []
        for i in range(0,6):
            if i == 0:
                base[i] = datetime.fromtimestamp(base[i]/1000)
            tmp.append(base[i])
        pair.append(tmp)
    df = pd.DataFrame(pair, columns=['date', 'open', 'high', 'low', 'close', 'volume'])
    df.date = pd.to_datetime(df.date)
    df.set_index("date", inplace=True)
    df = df.astype(float)
    return df

# # Financial indicators
def MA(df, period):
    return abstract.MA(df, timeperiod=period, matype=0)

if __name__ == "__main__":
    kline = GetKline(url, coin, '1d')
    index = MA(kline, 200)
    print (index)
利用幣安日線數據計算 200日均線

傳送到 Line Notify

# # 站上200日均線

import pandas as pd
import requests

from talib    import abstract
from datetime import datetime

# # DataFrame Setting
pd.set_option('display.max_columns', None)
pd.set_option('display.max_rows', None)
pd.set_option('max_colwidth',100)
pd.set_option('display.width', 5000)
pd.set_option('display.unicode.ambiguous_as_wide', True)
pd.set_option('display.unicode.east_asian_width', True)

# # Global Variables Setting
url  = 'https://api.binance.com/'
coin = 'BTCUSDT'

# # Line Notify
def send_ifttt(msg):   
    url = ('https://maker.ifttt.com/trigger/事件名稱/with/'+'key/金鑰'+'?value1='+str(msg))
    r = requests.get(url)      
    if r.text[:5] == 'Congr':  
        print('Success send (' +str(msg)+') to Line')
    return r.text

# # Get Market Data
def GetKline(url, symbol, interval):
    try:
        data = requests.get(url + 'api/v3/klines', params={'symbol': symbol, 'interval': interval, 'limit': 1000}).json()
    except Exception as e:
        print ('Error! problem is {}'.format(e.args[0]))
    tmp  = []
    pair = []
    for base in data:
        tmp  = []
        for i in range(0,6):
            if i == 0:
                base[i] = datetime.fromtimestamp(base[i]/1000)
            tmp.append(base[i])
        pair.append(tmp)
    df = pd.DataFrame(pair, columns=['date', 'open', 'high', 'low', 'close', 'volume'])
    df.date = pd.to_datetime(df.date)
    df.set_index("date", inplace=True)
    df = df.astype(float)
    return df

def GetAvgPrice(url, symbol):
    try:
        price = requests.get(url + 'api/v3/avgPrice', params={'symbol': symbol}).json()['price']
    except Exception as e:
        print ('Error! problem is {}'.format(e.args[0]))
    return float(price)

# # Financial indicators
def MA(df, period):
    return abstract.MA(df, timeperiod=period, matype=0)

if __name__ == "__main__":
    kline = GetKline(url, coin, '1d')
    index = MA(kline, 200)
    price = GetAvgPrice(url, coin)
    if price > index[-1]:
        print (f'當前價格為: {price}, 高於200日均線: {index[-1]}')
當前價格高於 200日均線

傳送至 WebHook

要呼叫 WebHook 讓行情資訊到 Line 之前，我們必須先至 IFTTT WebHook Setting 中獲取 IFTTTT 提供的 WebHook 網址。

WebHook URL 獲取

將 WebHook 網址輸入至瀏覽器並在其後加上 「 ?value1=測試」即可測試 Line Notify 是否正常運作。

WebHook 事件觸發
Line Notify 通知
# # Python 呼叫 WebHook

# # Line Notify
def Line(msg):   
    url = ('https://maker.ifttt.com/trigger/事件名稱/with/'+
          'key/金鑰' +
          '?value1='+str(msg))
    r = requests.get(url)      
    if r.text[:5] == 'Congr':  
        print('成功推送 (' +str(msg)+') 至 Line')
    return r.text

附上價格低於 200 MA 提醒完整程式碼

# # 200MA Alartor

import pandas as pd
import requests
import time

from talib    import abstract
from datetime import datetime

# # DataFrame Setting
pd.set_option('display.max_columns', None)
pd.set_option('display.max_rows', None)
pd.set_option('max_colwidth',100)
pd.set_option('display.width', 5000)
pd.set_option('display.unicode.ambiguous_as_wide', True)
pd.set_option('display.unicode.east_asian_width', True)

# # Global Variables Setting
url  = 'https://api.binance.com/'
coin = 'BTCUSDT'

# # Line Notify
def Line(msg):   
    url = ('https://maker.ifttt.com/trigger/事件名稱/with/'+
          'key/金鑰' +
          '?value1='+str(msg))
    r = requests.get(url)      
    if r.text[:5] == 'Congr':  
        print('成功推送 (' +str(msg)+') 至 Line')
    return r.text

# # Get Market Data
def GetKline(url, symbol, interval):
    try:
        data = requests.get(url + 'api/v3/klines', params={'symbol': symbol, 'interval': interval, 'limit': 1000}).json()
    except Exception as e:
        print ('Error! problem is {}'.format(e.args[0]))
    tmp  = []
    pair = []
    for base in data:
        tmp  = []
        for i in range(0,6):
            if i == 0:
                base[i] = datetime.fromtimestamp(base[i]/1000)
            tmp.append(base[i])
        pair.append(tmp)
    df = pd.DataFrame(pair, columns=['date', 'open', 'high', 'low', 'close', 'volume'])
    df.date = pd.to_datetime(df.date)
    df.set_index("date", inplace=True)
    df = df.astype(float)
    return df

def GetAvgPrice(url, symbol):
    try:
        price = requests.get(url + 'api/v3/avgPrice', params={'symbol': symbol}).json()['price']
    except Exception as e:
        print ('Error! problem is {}'.format(e.args[0]))
    return float(price)

# # Financial indicators
def MA(df, period):
    return abstract.MA(df, timeperiod=period, matype=0)

if __name__ == "__main__":
    while True:
        kline = GetKline(url, coin, '1d')
        index = MA(kline, 200)
        price = GetAvgPrice(url, coin)
        if price < index[-1]:
            msg = f'當前價格為: {price}, 低於200日均線: {index[-1]}'
            Line(msg)
        time.sleep(14400) # 4小時執行一次

小結

TA-LIB 套件提供大部分常見的技術指標運算方法，使用方式可以透過作者網站查詢。如果已經會利用 Python + IFTTT 推送 LINE 訊息後，就可以 用 TA-LIB 計算各種指標、判斷邏輯，在觸發某些事件時 ( 例如 RSI 過熱、DEMA 上穿、KDJ 金叉…等 ) 就會自動推送 Line 訊息通知，把握住每一次的交易機會。

此外 TradingView 也有提供付費會員 WebHook 推送的服務，如果各位客家人發財之後想要買會員省力，可以透過此推薦連結購買給筆者一點鼓勵。

TradingView 的 WebHook 提醒方式，填入 IFTTT WebHook 網址即可串接

如果對於文章內容、程式、環境安裝或者 IFTTT 有興趣的小夥伴也歡迎在下方留言與我討論喔！

加密貨幣屬於高風險投資，本網站內容均不構成任何投資建議與責任。

掌握虛擬貨幣、區塊鏈大小事

Grenade_LOGO_橫7

我們深耕台灣區塊鏈行業，擁有最活躍的社群，提供區塊鏈相關一站式服務。

Facebook-f Twitter Telegram Youtube Instagram Line

Grenade

區塊鏈資訊

交易所