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)