StockTrendAnalysis-LSTM

Sleeping

Update app.py

d1d2533 verified 6 months ago

7.67 kB

import gradio as gr
import sys
import time
import queue
import threading
import tensorflow as tf
import numpy as np
import matplotlib.pyplot as plt
import os
from sklearn import datasets
from sklearn.cluster import KMeans
from sklearn.cluster import DBSCAN
import pandas as pd
from sklearn.tree import DecisionTreeClassifier
from sklearn.ensemble import RandomForestClassifier
from sklearn.neighbors import KNeighborsClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
from sklearn.tree import export_text
from sklearn import metrics
import datetime
from sklearn.neural_network import MLPClassifier
from stock_manager import StockDataManager
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
manager = StockDataManager()


def run_all(tickersymbol, download_data, download_update, use_train, PREDICT, predict_date):
    # 產生設定參數摘要
    config_str = f"股票代碼：{tickersymbol}\n"
    config_str += f"更新資料：{'是' if download_data else '否'}{f' (日期: {download_update})' if download_data else ''}\n"
    config_str += f"啟用訓練：{'是' if use_train else '否'}\n"
    config_str += f"啟用預測：{'是' if PREDICT else '否'}{f'(日期: {predict_date})'if PREDICT else ''}\n"
    
    yield config_str


# --- 新增自訂的標準輸出攔截器 ---
class GlobalStreamInterceptor:
    def __init__(self, q, original_stdout):
        self.q = q
        self.original_stdout = original_stdout
        
    def write(self, buf):
        self.original_stdout.write(buf)
        # 這裡不 strip，保留原始內容給邏輯判斷
        if buf not in ["\n", ""]:
            self.q.put(buf)
        
    def flush(self):
        # 必須呼叫原始 stdout 的 flush，確保資料被清出緩衝區
        self.original_stdout.flush()
# 儲存最原始的 stdout
original_stdout = sys.stdout
stdout_lock = threading.Lock()

def run_external_code(tickersymbol, download_data, download_update, use_train, method, PREDICT, predict_date):
    q = queue.Queue()
        
    def background_task():
        # 取得鎖，確保不會有兩個任務同時搶奪 stdout
        with stdout_lock:
            old_stdout = sys.stdout
            sys.stdout = GlobalStreamInterceptor(q, old_stdout)
        try:
            # 執行原本的邏輯
            if download_data:
                manager.fetch_data(end_date=download_update)
            else:
                print("Populating ticker from files...")
                manager.load_raw_data(isProcessData=False)
                # reads CSV files saved by fetch_data()
            if use_train:
                manager.trainModel(
                ticker = tickersymbol,
                epochs_general = 10,
                epochs_particular = 10
            )
                #manager.trainMTAModel(ticker=tickersymbol, loss_fn=method)
                    
            if PREDICT:
                signal = manager.predict_day(ticker=tickersymbol, date=predict_date)
                print(f"\nFinal signal: {signal:+.4f}")
                
        except Exception as e:
            print(f"Error occurred: {str(e)}")
        finally:
            # 確保還原
            with stdout_lock:
                sys.stdout = original_stdout
            q.put(None)

    thread = threading.Thread(target=background_task, daemon=True)
    thread.start()
    log_messages = ""
    # 監聽佇列
    while True:
        try:
            # 嘗試取得佇列內容
            chunk = q.get(timeout=0.5)
            
            # 檢查是否為結束信號
            if chunk is None: 
                break
            
            clean_chunk = str(chunk).strip()
            if clean_chunk:
                log_messages += clean_chunk + "\n"
                yield log_messages
        
        # 修正 SyntaxError：補上 try 對應的 except 區塊
        except queue.Empty:
            # 如果佇列空了但執行緒已結束，則跳出
            if not thread.is_alive():
                break
            continue
            


def update_setup(is_checked):
    return gr.update(visible=is_checked)

def train_setup(is_checked):
    return gr.update(visible=is_checked)

def predict_setup(is_checked):
    return gr.update(visible=is_checked)

def update_ticker_list(market_choice):
    if market_choice == "TW50":
        if not manager.tw50_tickers:
            manager.fillTw50Tickers()
        new_choices = manager.tw50_tickers if manager.tw50_tickers else ["抓取TW50失敗"]
        return gr.update(choices=new_choices, value=new_choices[0])
        
    elif market_choice == "SP500":
        if not manager.sp500_tickers:
            manager.fillSp500Tickers()
        new_choices = manager.sp500_tickers if manager.sp500_tickers else ["抓取SP50失敗"]
        return gr.update(choices=new_choices, value=new_choices[0])
        
    return gr.update(choices=["No tickers loaded"], value="No tickers loaded")

symbol_list = ["No tickers loaded"]
today = datetime.date.today()
yesterday = today - datetime.timedelta(days=1)
tomorow = today + datetime.timedelta(days=1)


with gr.Blocks() as demo:
    gr.Markdown("### 股票分析與訓練設定")
    
    with gr.Row():
        market_radio = gr.Radio(
            choices = ["TW50", "SP500"], 
            label = "選擇市場板塊", 
            value = None
        )
        ticker_list = gr.Dropdown(
            choices = symbol_list, 
            label = "股票代碼",
            value = symbol_list[0]
        )
        
    with gr.Row():        
        UPDATE_DATA = gr.Checkbox(label="更新資料", value=False) 
        
        end_date = gr.DateTime(
            label = "選擇數據更新日",
            type = "string",       
            value = yesterday.strftime("%Y-%m-%d"), 
            interactive=True,
            include_time = False,
            visible = False
        ) 
        
    with gr.Row():    
        TRAIN = gr.Checkbox(label="啟用訓練功能", value=False) 
        
        
    with gr.Row():    
        PREDICT = gr.Checkbox(label="啟用預測功能", value=False) 
        
        predict_date = gr.DateTime(
            label = "選擇預測日期",
            type = "string",       
            value = today.strftime("%Y-%m-%d"), 
            interactive=True,
            include_time = False,
            visible = False
        ) 
   
    with gr.Row():
        run_btn = gr.Button("啟動", variant="primary")
        
    ##--觸發事件--
    market_radio.change(
        fn = update_ticker_list,
        inputs = market_radio,
        outputs = ticker_list
    )

    UPDATE_DATA.change(
        fn = update_setup, 
        inputs = UPDATE_DATA, 
        outputs = end_date
    )
        
    PREDICT.change(
        fn = predict_setup, 
        inputs = PREDICT, 
        outputs = predict_date
    )
  
    #--兩個輸出欄位--
    config_display = gr.Textbox(label="目前設定參數", lines=3)

    # 加上 autoscroll=True 讓日誌自動向下捲動
    output_text = gr.Textbox(label="控制台輸出內容", lines=10, autoscroll=True)

    # 第一個 Click 事件負責更新左邊/上方的參數區塊
    run_btn.click(
        fn = run_all, 
        inputs = [ticker_list, UPDATE_DATA, end_date, TRAIN, PREDICT, predict_date], 
        outputs = [config_display] 
    )
    # 第二個 Click 事件負責在背景跑訓練，並透過 yield 即時更新 output_text
    run_btn.click(
        fn = run_external_code,
        inputs = [ticker_list, UPDATE_DATA, end_date, TRAIN, PREDICT, predict_date],
        outputs = output_text
    )


demo.launch(ssr_mode=False)

	import gradio as gr
	import sys
	import time
	import queue
	import threading
	import tensorflow as tf
	import numpy as np
	import matplotlib.pyplot as plt
	import os
	from sklearn import datasets
	from sklearn.cluster import KMeans
	from sklearn.cluster import DBSCAN
	import pandas as pd
	from sklearn.tree import DecisionTreeClassifier
	from sklearn.ensemble import RandomForestClassifier
	from sklearn.neighbors import KNeighborsClassifier
	from sklearn.model_selection import train_test_split
	from sklearn.metrics import accuracy_score
	from sklearn.tree import export_text
	from sklearn import metrics
	import datetime
	from sklearn.neural_network import MLPClassifier
	from stock_manager import StockDataManager
	os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
	manager = StockDataManager()


	def run_all(tickersymbol, download_data, download_update, use_train, PREDICT, predict_date):
	# 產生設定參數摘要
	config_str = f"股票代碼：{tickersymbol}\n"
	config_str += f"更新資料：{'是' if download_data else '否'}{f' (日期: {download_update})' if download_data else ''}\n"
	config_str += f"啟用訓練：{'是' if use_train else '否'}\n"
	config_str += f"啟用預測：{'是' if PREDICT else '否'}{f'(日期: {predict_date})'if PREDICT else ''}\n"

	yield config_str


	# --- 新增自訂的標準輸出攔截器 ---
	class GlobalStreamInterceptor:
	def __init__(self, q, original_stdout):
	self.q = q
	self.original_stdout = original_stdout

	def write(self, buf):
	self.original_stdout.write(buf)
	# 這裡不 strip，保留原始內容給邏輯判斷
	if buf not in ["\n", ""]:
	self.q.put(buf)

	def flush(self):
	# 必須呼叫原始 stdout 的 flush，確保資料被清出緩衝區
	self.original_stdout.flush()
	# 儲存最原始的 stdout
	original_stdout = sys.stdout
	stdout_lock = threading.Lock()

	def run_external_code(tickersymbol, download_data, download_update, use_train, method, PREDICT, predict_date):
	q = queue.Queue()

	def background_task():
	# 取得鎖，確保不會有兩個任務同時搶奪 stdout
	with stdout_lock:
	old_stdout = sys.stdout
	sys.stdout = GlobalStreamInterceptor(q, old_stdout)
	try:
	# 執行原本的邏輯
	if download_data:
	manager.fetch_data(end_date=download_update)
	else:
	print("Populating ticker from files...")
	manager.load_raw_data(isProcessData=False)
	# reads CSV files saved by fetch_data()
	if use_train:
	manager.trainModel(
	ticker = tickersymbol,
	epochs_general = 10,
	epochs_particular = 10
	)
	#manager.trainMTAModel(ticker=tickersymbol, loss_fn=method)

	if PREDICT:
	signal = manager.predict_day(ticker=tickersymbol, date=predict_date)
	print(f"\nFinal signal: {signal:+.4f}")

	except Exception as e:
	print(f"Error occurred: {str(e)}")
	finally:
	# 確保還原
	with stdout_lock:
	sys.stdout = original_stdout
	q.put(None)

	thread = threading.Thread(target=background_task, daemon=True)
	thread.start()
	log_messages = ""
	# 監聽佇列
	while True:
	try:
	# 嘗試取得佇列內容
	chunk = q.get(timeout=0.5)

	# 檢查是否為結束信號
	if chunk is None:
	break

	clean_chunk = str(chunk).strip()
	if clean_chunk:
	log_messages += clean_chunk + "\n"
	yield log_messages

	# 修正 SyntaxError：補上 try 對應的 except 區塊
	except queue.Empty:
	# 如果佇列空了但執行緒已結束，則跳出
	if not thread.is_alive():
	break
	continue



	def update_setup(is_checked):
	return gr.update(visible=is_checked)

	def train_setup(is_checked):
	return gr.update(visible=is_checked)

	def predict_setup(is_checked):
	return gr.update(visible=is_checked)

	def update_ticker_list(market_choice):
	if market_choice == "TW50":
	if not manager.tw50_tickers:
	manager.fillTw50Tickers()
	new_choices = manager.tw50_tickers if manager.tw50_tickers else ["抓取TW50失敗"]
	return gr.update(choices=new_choices, value=new_choices[0])

	elif market_choice == "SP500":
	if not manager.sp500_tickers:
	manager.fillSp500Tickers()
	new_choices = manager.sp500_tickers if manager.sp500_tickers else ["抓取SP50失敗"]
	return gr.update(choices=new_choices, value=new_choices[0])

	return gr.update(choices=["No tickers loaded"], value="No tickers loaded")

	symbol_list = ["No tickers loaded"]
	today = datetime.date.today()
	yesterday = today - datetime.timedelta(days=1)
	tomorow = today + datetime.timedelta(days=1)


	with gr.Blocks() as demo:
	gr.Markdown("### 股票分析與訓練設定")

	with gr.Row():
	market_radio = gr.Radio(
	choices = ["TW50", "SP500"],
	label = "選擇市場板塊",
	value = None
	)
	ticker_list = gr.Dropdown(
	choices = symbol_list,
	label = "股票代碼",
	value = symbol_list[0]
	)

	with gr.Row():
	UPDATE_DATA = gr.Checkbox(label="更新資料", value=False)

	end_date = gr.DateTime(
	label = "選擇數據更新日",
	type = "string",
	value = yesterday.strftime("%Y-%m-%d"),
	interactive=True,
	include_time = False,
	visible = False
	)

	with gr.Row():
	TRAIN = gr.Checkbox(label="啟用訓練功能", value=False)


	with gr.Row():
	PREDICT = gr.Checkbox(label="啟用預測功能", value=False)

	predict_date = gr.DateTime(
	label = "選擇預測日期",
	type = "string",
	value = today.strftime("%Y-%m-%d"),
	interactive=True,
	include_time = False,
	visible = False
	)

	with gr.Row():
	run_btn = gr.Button("啟動", variant="primary")

	##--觸發事件--
	market_radio.change(
	fn = update_ticker_list,
	inputs = market_radio,
	outputs = ticker_list
	)

	UPDATE_DATA.change(
	fn = update_setup,
	inputs = UPDATE_DATA,
	outputs = end_date
	)

	PREDICT.change(
	fn = predict_setup,
	inputs = PREDICT,
	outputs = predict_date
	)

	#--兩個輸出欄位--
	config_display = gr.Textbox(label="目前設定參數", lines=3)

	# 加上 autoscroll=True 讓日誌自動向下捲動
	output_text = gr.Textbox(label="控制台輸出內容", lines=10, autoscroll=True)

	# 第一個 Click 事件負責更新左邊/上方的參數區塊
	run_btn.click(
	fn = run_all,
	inputs = [ticker_list, UPDATE_DATA, end_date, TRAIN, PREDICT, predict_date],
	outputs = [config_display]
	)
	# 第二個 Click 事件負責在背景跑訓練，並透過 yield 即時更新 output_text
	run_btn.click(
	fn = run_external_code,
	inputs = [ticker_list, UPDATE_DATA, end_date, TRAIN, PREDICT, predict_date],
	outputs = output_text
	)


	demo.launch(ssr_mode=False)