Local_Stock_Line_Notify.py

資料清洗完後，透過 VIEW 取得最終結果，將開盤價、最高價、最低價、漲幅等資訊繪製成折線圖，存檔之後透過 API 發送至 LINE Notify。

Local_Stock_Line_Notify.py

from datetime import datetime, timedelta
import requests
import datetime
from datetime import datetime
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
import matplotlib.colors as mcolors
# 以下 package 需事先使用 pip install 安裝
import pandas
import pyodbc

'''
==================================================================================================================
Author: Skylar Ma
------------------------------------------------------------------------------------------------------------------
Description: Get the stock price information and draw it into a line chart and send it through LINE Notify.
------------------------------------------------------------------------------------------------------------------
Version  Date        Note
1.0      2023-05-27  Init.
==================================================================================================================
'''

# For MS-SQL
TABLE = "vw_For_Line_Notify"


def send_line_notify(symbolId, now, line_token, pic_folder):
    url = "https://notify-api.line.me/api/notify"
    headers = {
        "Authorization": f"Bearer {line_token}"
    }
    payload = {
        "message": f"{symbolId} 當日股價資訊"
    }
    image = open(f'{pic_folder}\\{symbolId}_{now}.png', 'rb')    # 以二進位方式開啟圖片
    imageFile = {'imageFile' : image}   # 設定圖片資訊
    r = requests.post(url, headers=headers, params=payload, files=imageFile)
    return r.status_code

def query_table(server_name, database, account, pwd, line_token, pic_folder):
    # 開啟 SQL Server 連線
    cnxn = pyodbc.connect('DRIVER={SQL Server};SERVER='+server_name+';DATABASE='+database+';UID='+account+';PWD='+ pwd)
    cursor = cnxn.cursor()

    # 查詢現在關注哪些標的
    query = "SELECT [stock_id] FROM [Stock_API].[dbo].[Stock_Get] WHERE isGet = 'TRUE';"
    df = pandas.read_sql(query, cnxn)

    for index, item in df.iterrows():
        symbolId = item['stock_id']
        # 設定前後七天
        to_date = datetime.today()
        from_date = to_date - timedelta(days=7)
        to_date_str = to_date.strftime("%Y-%m-%d")
        from_date_str = from_date.strftime("%Y-%m-%d")
        # 設定 Query 語法
        query_status_1 = f"SELECT * FROM [Stock_API].[dbo].[vw_For_Line_Notify] WHERE FullDate BETWEEN '{from_date_str}' AND '{to_date_str}' AND symbolId = '{symbolId}' AND SRC = 'Today';"
        # 讀取資料並寫入 Dataframe 中
        df1 = pandas.read_sql(query_status_1, cnxn)

        # 建立一個圖形，裡面包含四個子圖表，並設定大小為 8*6
        fig, (ax1, ax2, ax3, ax4) = plt.subplots(4, constrained_layout=True, figsize=(8, 6))

        # 【第一個子圖表：開盤價】
        ax1.plot(df1['date'], df1['priceOpen'], color = 'lightcoral', marker='o', markersize=4, label='Opening')
        # 取得最後一個資料點的座標
        last_date = df1['date'].iloc[-1]
        last_price = df1['priceOpen'].iloc[-1]
        # 在最後一個資料點上標上資料文字
        ax1.text(last_date, last_price, f'{last_price}', ha='right')

        # 【第二個子圖表：當日最高價】
        ax2.plot(df1['date'], df1['priceHigh'], color = 'sandybrown', marker='o', markersize=4, label='High')
        # 取得最後一個資料點的座標
        last_date = df1['date'].iloc[-1]
        last_price = df1['priceHigh'].iloc[-1]
        # 在最後一個資料點上標上資料文字
        ax2.text(last_date, last_price, f'{last_price}', ha='right')

        # 【第三個子圖表：當日最低價】
        ax3.plot(df1['date'], df1['priceLow'], color = 'mediumaquamarine', marker='o', markersize=4, label='Low')
        # 取得最後一個資料點的座標
        last_date = df1['date'].iloc[-1]
        last_price = df1['priceLow'].iloc[-1]
        # 在最後一個資料點上標上資料文字
        ax3.text(last_date, last_price, f'{last_price}', ha='right')

        # 【第四個子圖表：漲跌幅】
        ax4.plot(df1['date'], df1['changePercent'], color = 'steelblue', marker='o', markersize=4, label='Change\nPercent')
        # 取得最後一個資料點的座標
        last_date = df1['date'].iloc[-1]
        last_price = df1['changePercent'].iloc[-1]
        # 在最後一個資料點上標上資料文字
        ax4.text(last_date, last_price, f'{last_price}', ha='right')

        # 加上標題與座標名稱
        ax1.set_title(f"{symbolId} Opening")
        ax1.set_xlabel("Date")
        ax1.set_ylabel("Price")

        ax2.set_title(f"{symbolId} High")
        ax2.set_xlabel("Date")
        ax2.set_ylabel("Price")

        ax3.set_title(f"{symbolId} Low")
        ax3.set_xlabel("Date")
        ax3.set_ylabel("Price")

        ax4.set_title(f"{symbolId} Change Percent")
        ax4.set_xlabel("Date")
        ax4.set_ylabel("%")

        # 加上圖例與位置調整
        ax1.legend(bbox_to_anchor=(1.05, 1.0), loc='upper left')
        ax2.legend(bbox_to_anchor=(1.05, 1.0), loc='upper left')
        ax3.legend(bbox_to_anchor=(1.05, 1.0), loc='upper left')
        ax4.legend(bbox_to_anchor=(1.05, 1.0), loc='upper left')

        # 調整子圖間距
        fig.tight_layout()

        # 儲存圖表
        now = datetime.today().strftime('%Y%m%d%H%M%S%f')[:-3]
        plt.savefig(f'{pic_folder}\\{symbolId}_{now}.png',
                transparent=False,
                bbox_inches='tight',
                pad_inches=1)

        # 顯示圖形（測試用）
        #plt.show()

        status_code = send_line_notify(symbolId, now, line_token, pic_folder)
        if status_code == 200:
            print("訊息傳送成功！")
        else:
            print("訊息傳送失敗，錯誤碼：", status_code)
            
    cursor.close()
    cnxn.close()



def convertDatetime(time_string):
    # 解析時間字符串
    datetime_obj = datetime.fromisoformat(time_string)
    date_str = datetime_obj.strftime('%H:%M:%S')

    return date_str

# 測試用
#query_table(server_name, database, account, pwd, line_token, pic_folder)