技術的な話

Google Financeの値をPythonでスクレイピングしてSlackに通知する

意識高く毎朝所有している株価の価格や経済指標値がSlackに通知されると良い気がして思わず作りました。普通に株アプリとかを立ち上げても良いのですが、他の確認と共に出来ると一石二鳥です。（面倒臭いというのは内緒。）

PythonでスクレイピングするのはBeautifulSoupを使うのがポピュラー？な気がしますが、今回作成する機能は非常にシンプルなので、そこまで機能的に必要無かったのでlxmlを使用しました。

そもそもスクレイピングしてOK？

ここは最初に確認する必要があります。

https://www.google.com/robots.txtにアクセスして調べます。他のサイトをスクレイピングする場合も同様にチェックしましょう。

今回スクレイピング対象となるhttps://www.google.com/finance/は許可されていそうです。

# robots.txt抜粋

Disallow: /maps/reserve/partner-dashboard
Disallow: /about/views/
Disallow: /intl/*/about/views/
Disallow: /local/cars
Disallow: /local/cars/
Disallow: /local/dealership/
Disallow: /local/dining/
Disallow: /local/place/products/
Disallow: /local/place/reviews/
Disallow: /local/place/rap/
Disallow: /local/tab/
Disallow: /localservices/*
Allow: /finance
Allow: /js/
Disallow: /nonprofits/account/
Disallow: /fbx
Disallow: /partners/agency?id=*

さくっと実装する

流れとしてはざっくり以下になります。

  1. 必要なライブラリをインストールします。
    pip install lxml
  2. 対象の経済指標値をURLに含めてrequests経由でアクセスします。
  3. 今日と昨日終値を画面から取得します。（下記画像の赤枠部分）
  4. 2.で取得した値から変化率や変化値を計算して取得します。
    ※これも画面から取得すれば良かったのですが、うまいこと取れなかったので計算してます。
  5. Slackに通知を送信します。

IdもしくはClassで要素取得できれば良かったのですが、結果的にXPATHで要素取得した方が楽だったのでそうしました。XPATHは変わる可能性があるので上手く動作しない場合はChromeの開発者機能を使用してXPATHがどうなっているかを確認する必要があります。

SLACK_URL部分にはSlack IncomingWebhookのURLを記載してください。

ちなみにここに記載しているコードをそのままコピーすると下記値を通知してくれます。

  • ダウ平均株価
  • ナスダック総合指数
  • S&P 500
  • VIX
  • REIT
  • 日経平均株価

import os
import json
import random
import requests
from lxml import html
from typing import Dict, Tuple
from datetime import datetime
from logging import getLogger, Logger, basicConfig, INFO

basicConfig(level=INFO)
logger:Logger = getLogger()

# slack url
SLACK_URL:str = "<your-slack-url>"

USER_AGENT:str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36"
REQUEST_HEADER:Dict = {"User-Agent": USER_AGENT}

CURRENT_VALUE_XPATH:str = "//*[@id='yDmH0d']/c-wiz/div/div[4]/div/div/main/div[2]/div[1]/div[1]/c-wiz/div/div[1]/div/div[1]/div/div[1]/div/span/div/div"
PREVIOUS_VALUE_XPATH:str = "//*[@id='yDmH0d']/c-wiz/div/div[4]/div/div/main/div[2]/div[2]/div/div[1]/div[2]/div"

def remove_commas(source: str) -> str:
    return source.replace(",", "")

def scrayping_by_economic_indicator(economic_indicator: str) -> Tuple[float, float, float, float]:
    """対象ワードを検索して値を取得
    Args:
        economic_indicator (str): 対象経済指標値
        Google Finance (https://www.google.com/finance/) にアクセスして、
        調査したい指標値を参照し、URLの末尾をeconomic_indicatorに指定してください
    """
    res = requests.get(f"https://www.google.com/finance/quote/{economic_indicator}", headers=REQUEST_HEADER)

    lxml_parsing = html.fromstring(res.text)

    # current values
    current_values = lxml_parsing.xpath(CURRENT_VALUE_XPATH)
    # previous value
    previous_values = lxml_parsing.xpath(PREVIOUS_VALUE_XPATH)

    if len(current_values) == 0:
        raise Exception(f"element not found. economic_indicator: {economic_indicator}")
    
    if len(current_values) > 1:
        logger.warning(f"multiple element detected. return first element.")

    current_value:float = float(remove_commas(current_values[0].text))
    previous_value:float = float(remove_commas(previous_values[0].text))
    
    change_value:float = round(float(previous_value) - float(current_value), 2)
    change_rate:float = round(change_value/float(previous_value) * 100, 2)
    logger.info(f"[caluculate result][{economic_indicator}] current value: {current_value} / previous_value: {previous_value} / change_value: {change_value} / change_rate: {change_rate}")
    
    return (current_value, previous_value, change_value, change_rate)

def notify_slack(economic_indicator_name:str, current_value:float, previous_value:float, change_value:float, change_rate:float) -> None:
    """SLACK_URLに対して通知を送信します
    Args:
        economic_indicator_name (str): 経済指標値名称
        current_value (float): 今日の値
        previous_value (float): 昨日の値
        change_value (float): 変化値
        change_rate (float): 変化率
    """
    rate_simbol:str = "+" if change_value >= 0 else "-"
    format_date:str = datetime.now().strftime('%Y/%m/%d')
    requests.post(SLACK_URL, data=json.dumps({
        "text" : f"{format_date}\n経済指標値： *{economic_indicator_name}* \n昨日値：{previous_value}\n本日値：{current_value}（{rate_simbol}{abs(change_rate)}）",
        "username" : "economic_indicators_bot"
    }))

def main() -> None:
    logger.info("economic indicators scrayping start.")

    notify_slack("DOW", *scrayping_by_economic_indicator(".DJI:INDEXDJX"))
    notify_slack("NASDAQ", *scrayping_by_economic_indicator(".IXIC:INDEXNASDAQ"))
    notify_slack("S&P500", *scrayping_by_economic_indicator(".INX:INDEXSP"))
    notify_slack("VIX", *scrayping_by_economic_indicator("VIX:INDEXCBOE"))
    notify_slack("REIT指数", *scrayping_by_economic_indicator("REIT:INDEXTYO"))
    notify_slack("日経平均株価", *scrayping_by_economic_indicator("NI225:INDEXNIKKEI"))
    
    logger.info("economic indicators scrayping finish.")

if __name__ == "__main__":
    main()

手動で実行して問題無くSlackに通知が来たら後はcron等に登録して自動実行させます。

私の場合は/etc/crontabに下記を追記して自動実行設定をしました。Windowsの場合はタスクスケジューラとかでも良いかと思います。

# Example of job definition:
# .---------------- minute (0 - 59)
# |  .------------- hour (0 - 23)
# |  |  .---------- day of month (1 - 31)
# |  |  |  .------- month (1 - 12) OR jan,feb,mar,apr ...
# |  |  |  |  .---- day of week (0 - 6) (Sunday=0 or 7) OR sun,mon,tue,wed,thu,fri,sat
# |  |  |  |  |
# *  *  *  *  * user-name command to be executed 
5  8  *  *  *   root    python main.py >> app.log 2>&1

こんな感じで毎朝通知が来るようになります。

経済に明るい社会人でありたいものです。

-技術的な話
-, , , ,