用 Python 建構 Telegram 股票監測機器人

Posted on Edited on In Disqus:

有投資美股的人應該都有一樣的經歷，因為美國跟台灣的時差12小時，看盤可能都得要通宵坐鎮。因為我實在太愛睡覺了，年紀越大也越經不起熬夜折磨，所以就設法用最簡單的方式完成投資時需要做的事情。

為了讓解決方案能夠滿足我的需求，我思考了需要具備的特性：

  • 能隨時隨地查詢
  • 能主動告警
  • 客製化圖表
  • 未來可擴充 ML/DL 算法

經過摸索後，Telegram bots 可以很簡單快速的達到我的需求，而且彈性且自由。因此這篇主要會著重在如何架構出 Telegram 機器人，並幫助我達到最基本的需求。後續我會再不斷優化這個機器人，有興趣的可以關注我的Github。

本篇文章重點:

  • 註冊 Telegram bots
  • 用 Python 操作 Telegram bots
  • 架構互動選單
  • 股價波動告警

註冊 Telegram bots

要使用 Telegram 的機器人其實並不困難，官方提供的文件就足以建立自己的機器人。這裡我們手把手把流程跑一次。

首先，用自己的 Telegram 新增好友 BotFather，會出現一個很帥的機器人大叔。

圖片

加入他後輸入 /start 就可以取得建立一個機器人所需要的指令。這裡我們得知建立機器人輸入指令/newbot，並依序輸入顯示名稱以及使用者帳號（用來搜尋以及呼叫用的）。

圖片

只要名稱帳號沒有重複，你會看到以下成功訊息，就代表你的機器人建好囉！基本上現在你去搜尋好友的地方打上你剛剛輸入的使用者名稱，就可以找到你的機器人。不過你的機器人現在還是空殼，沒有駕駛人坐在裡面。你要能駕駛它，就需要用到 BotFather 提供的鑰匙（就是下面紅色那一塊 API ）。要特別注意的是，只要誰有這個鑰匙，誰就可以操作你的機器人，所以要格外小心這個鑰匙外流。

圖片

用 Python 操作 Telegram bots

到這一步我們就可以使用 Python 的套件 telegram 來連結我們建立好的機器人了。根據 官方文件 我們知道需要用 Updater 及 dispatcher 來接收及傳輸資訊。

1
2
3
4
5
6
import telegram
from telegram.ext import Updater

updater = Updater(token=’API 鑰匙‘, use_context=True)

dispatcher = updater.dispatcher

我們簡單設定一個歡迎語，確定我們有連結到機器人。這個歡迎語會在你第一次加入機器人為好友時發送。如想要手動啟動這個歡迎語，也可以直接在與機器人的對話框中輸入 /start。

1
2
3
4
5
6
7
8
def start(update, context):
    # 傳送訊息給使用者
context.bot.send_message(chat_id=update.effective_chat.id,
 text=”歡迎使用股市貓頭鷹🦉“)

from telegram.ext import CommandHandler
start_handler = CommandHandler(’start‘, start)
dispatcher.add_handler(start_handler)  # 把此 Handler 加入派送任務中

這個 CommandHandler 是接受指令的一個物件，當收到使用者傳送 / 開頭的指令內容時，就會交由這個 CommandHandler 來處理。Telegram 在新增機器人為好友時，就會送一個 /start 給機器人，這也是為什麼用戶加入時會收到這個歡迎語。

除了 CommandHandler 之外，還有 MessageHandler, ConversationHandler…等，用來處理使用者發送出的不同情境，詳細可以看官方網站做出不同變化。

到這裡還不算完成，直到執行下述指令。

1
2
updater.start_polling()  # 開始推送任務
# updater.stop()  # 停止推送任務

這時候再去 Telegram 中加入機器人，就可以看到你設定的歡迎語了。

架構互動選單

為了讓我們的機器人更厲害一點點，這裡來架設互動式的選單，讓他可以富含不同的功能。看起來就像這樣：

圖片

要做到選單功能，底層的流程架構必須借助 ConversationHandler ，我草畫了這個函數的功用，並不會很難理解。

簡單的用下圖來說明，一開始會由 start 函數開場，在使用者做出動作後，這個動作 (CHOOSING) 會傳至 ConversationHandler，並依據使用者的動作內容進行判斷，看接下來應該由哪個函數接手後續動作。以這個例子來說，ConversationHandler 判斷並將流程交給 check_and_store 函數，而後這個函數完成後，會再將結果傳給 COMFIRM_STOCK 進行判斷。依此不斷循環。

圖片

了解架構後，讓我們直接進入程式的部分。首先，我們先把功能選單的介面用出來。

1
2
3
4
5
6
7
8
9
10
# 載入功能選單的函數
from telegram import ReplyKeyboardMarkup

# 設定選單項目及名稱，此為一個二維矩陣
reply_keyboard = [['輸入/移除追蹤股票', '追蹤標的清單'],
                  ['查詢技術線圖'],
                  ['結束']]

# 存成選單物件
markup = ReplyKeyboardMarkup(reply_keyboard)

接下來，稍微修改一下剛剛的 start 函數。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 建立回應變數，就是剛剛架構裡所介紹的東西
# 裡面的值是什麼不重要，只要不重複就好
# 後續會用在 ConversationHandler 做字典的 key 值
CHOOSING, COMFIRM_STOCK, TYPING_CHOICE = range(3)

# 調整剛剛的 start 函數
# 主要改變的地方有兩個：
# 1. 將功能選單掛載上去
# 2. 新增回傳值 CHOOSING
def start(update, context):
	context.bot.send_message(chat_id=update.effective_chat.id, 
                             text='''
                             歡迎使用股市貓頭鷹🦉，請選擇你想要執行的服務，我會盡全力協助你的！
                             ''',
                             reply_markup=markup) # 功能選單
    return CHOOSING

函數的型態大致就如上面的 start 相同，因此我們先來介紹 ConverstationHandler 該如何設置。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from telegram.ext import (MessageHandler, Filters,
                          ConversationHandler)

conv_handler = ConversationHandler(
        entry_points=[CommandHandler('start', start)],

        states={
            CHOOSING: [MessageHandler(Filters.regex('^輸入/移除追蹤股票$') & ~(Filters.command | Filters.regex('^結束$')),
                                      input_stock),
                       MessageHandler(Filters.regex('^查詢技術線圖$') & ~(Filters.command | Filters.regex('^結束$')),
                                      get_k_plot),
                       MessageHandler(Filters.regex('^追蹤標的清單$') & ~(Filters.command | Filters.regex('^結束$')),
                                      check_stock_list),
                       MessageHandler(~(Filters.command | Filters.regex('^結束$')),
                                      wrong_command)
                       ],

            TYPING_CHOICE: [
                MessageHandler(Filters.regex('^[A-z0-9]*$') & ~(Filters.command | Filters.regex('^[qQ]$')),
                               check_and_store),
                MessageHandler(Filters.regex('^-[A-z0-9]*$') & ~(Filters.command | Filters.regex('^[qQ]$')),
                               stock_remove),
                MessageHandler(Filters.regex('^q$') & ~(Filters.command),
                               start)],

            COMFIRM_STOCK: [
                MessageHandler(Filters.text & ~(Filters.command | Filters.regex('^[qQ]$')),
                               stock_added),
                MessageHandler(Filters.regex('^[qQ]$') & ~(Filters.command),
                               start)],
        },

        fallbacks=[MessageHandler(Filters.regex('^結束$'), done)]
    )

這裡分幾個部分說明：

  • entry_points 是進到這個 ConversationHandler 的入口，也就是我們剛剛所建立的 start 函數，可以注意的是 start 函數中使用者輸入的結果會送到 CHOOSING 處理。
  • states 就是剛剛架構圖的主軸，是字典的資料型態，key 值就是我們剛剛設定的 CHOOSING, COMFIRM_STOCK, TYPING_CHOICE 等。
  • MessageHandler 跟 CommandHandler 功能差不多，但因為使用者在點選選單時，回傳的結果會是 Message 的型態，因此採用 MessageHandler 處理。要特別注意的是第二個參數會放入接著處理流程的函數，舉例來說，在 start 函數中使用者點選 輸入/移除追蹤股票 選項，那麼第二個參數就會放入 input_stock 這個函數來處理後續流程。
  • MessageHandler 第一個參數放入判斷值，這裡使用 Filters 來做判別，以 MessageHandler(Filters.regex('^輸入/移除追蹤股票$') & ~(Filters.command | Filters.regex('^結束$')), input_stock) 舉例，用正則表達式判別為 輸入/移除追蹤股票 且不是指令或是 結束 ，則進到下一個步驟 input_stock。
  • fallbacks 是結束這個 ConverstationHandler 執行的函數。

完成這個 ConversationHandler 物件的設定後，我們就可以把他加到我們的機器人中了。

1
dispatcher.add_handler(conv_handler)

如果想要多瞭解其他的函數及關係，可以參考我的 GitHub，裡面有更詳盡的代碼。

股價波動告警

有了以上的選單，機器人就可以依據我們的要求來回應結果了。但如果希望機器人不是單純的接受/回傳訊息，而是能主動根據不同情境跳出通知來警示我們，這樣就更有價值了。因此，這裡會實作如何讓機器人在特定時間或特定事件發生時跳出通知。

要使用這個功能，就必須借助 JobQueue 來實作。

1
2
# 建立 JobQueue 物件
job = updater.job_queue

這個 JobQueue 分別有以下幾種功能

  • job.run_once
  • job.run_repeating
  • job.run_daily
  • job.run_monthly

照著字面上的意思不難了解功用為何，我們來實作一個例子來幫助了解。這裡我們要建立一個提醒通知，每分鐘都會提醒現在過了幾分鐘。

1
2
3
4
5
6
7
8
9
10
11
# 初始化分鐘數
minute = 0

# 建立傳送訊息的函數
def count_minute(context: telegram.ext.CallbackContext):
    minute += 1
    context.bot.send_message(chat_id='傳送通知的用戶 ID',
                             text='現在過了{}分鐘囉！'.format(minute))

# 建立時間提醒 JobQueue 並執行
job_count_minute = job.run_repeating(count_minute, interval=60, first=0)

如果不是 job.run_once ，我們必須手動關閉該工作，可以透過以下指令來完成。

1
2
3
4
5
# 暫時停用該工作
job_count_minute.enabled = False  

# 永久刪除該工作
job_count_minute.schedule_removal()

透過上述函數，就可以達到每分鐘跳出提醒。依此類推，我們也可以設置每小時的檢查 JobQueue ，關注的股票有無最新新聞，如果有的話就跳通知，沒有的話則略過；或是建立一個在晚上期間關注股票變動超過 10% 時跳出告警等功能。

總結

透過上述對 telegram 套件的說明，我們簡單的介紹了如何用 Python 來架設 telegram 機器人，以幫助我們進一步追蹤股票或是應用在其他生活面向。因為隨著功能越多，設置的架構就會越複雜，在這篇文章就不多著墨在這些部分，有興趣的可以關注我的 GitHub。

如果你覺得這篇文章對你有幫助，還請熱情給我按個讚😂。期望你能順利製作出屬於你的機器人！