Create app.py

app.py

@@ -0,0 +1,175 @@
+# app.py
+import gradio as gr
+import yfinance as yf
+import pandas as pd
+import numpy as np
+from sklearn.preprocessing import MinMaxScaler
+from sklearn.pipeline import make_pipeline
+from sklearn.linear_model import Ridge
+from loguru import logger
+import time
+import threading
+import plotly.graph_objs as go
+
+# 配置日志
+logger.add("app.log", rotation="1 MB", level="DEBUG")
+
+def quick_feature_engineering(df):
+    """快速特征工程"""
+    df = df.copy()
+    # 基础特征
+    df['Returns'] = df['Close'].pct_change()
+    df['Volatility'] = df['Returns'].rolling(5).std()
+    # 简化时间特征
+    df['Day'] = df.index.dayofweek
+    df['Month'] = df.index.month
+    return df.dropna()
+
+def rapid_training(ticker):
+    """快速训练流程（必须在30秒内完成）"""
+    start_time = time.time()
+    
+    try:
+        # 获取数据（限制为1年数据）
+        logger.info(f"Fetching data for {ticker}")
+        data = yf.download(ticker, period="1y", progress=False)
+        if data.empty:
+            raise ValueError("No data available")
+        
+        # 特征工程
+        logger.debug("Processing features")
+        data = quick_feature_engineering(data)
+        
+        # 准备训练数据
+        X = data.drop(columns=['Close'])
+        y = data['Close']
+        
+        # 最后7天作为测试集
+        train_size = -7
+        X_train, y_train = X.iloc[:train_size], y.iloc[:train_size]
+        
+        # 使用轻量级模型管道
+        model = make_pipeline(
+            MinMaxScaler(),
+            Ridge(alpha=1.0)  # 调整正则化强度
+        
+        logger.info("Start training")
+        model.fit(X_train, y_train)
+        
+        # 生成预测（未来7天）
+        logger.debug("Generating predictions")
+        last_features = X.iloc[-1:].values
+        future_dates = pd.date_range(data.index[-1], periods=8)[1:]
+        predictions = []
+        
+        # 递归预测
+        current_features = last_features.copy()
+        for _ in range(7):
+            pred = model.predict(current_features)[0]
+            predictions.append(pred)
+            # 更新特征（简化处理）
+            current_features[0][0] = pred  # 更新Open
+            current_features[0][3] = pred  # 更新Close
+            
+        training_time = time.time() - start_time
+        logger.success(f"Training completed in {training_time:.2f}s")
+        
+        return {
+            'data': data,
+            'model': model,
+            'predictions': pd.Series(predictions, index=future_dates),
+            'training_time': training_time
+        }
+    
+    except Exception as e:
+        logger.error(f"Error in training: {str(e)}")
+        return None
+
+def create_plot(result):
+    """创建交互式图表"""
+    data = result['data']
+    pred = result['predictions']
+    
+    fig = go.Figure()
+    
+    # 历史价格
+    fig.add_trace(go.Scatter(
+        x=data.index,
+        y=data['Close'],
+        name='Historical Price',
+        line=dict(color='blue')
+    )
+    
+    # 预测价格
+    fig.add_trace(go.Scatter(
+        x=pred.index,
+        y=pred.values,
+        name='Prediction',
+        line=dict(color='red', dash='dot')
+    )
+    
+    fig.update_layout(
+        title=f"Stock Price Prediction",
+        xaxis_title='Date',
+        yaxis_title='Price (USD)',
+        hovermode="x unified",
+        showlegend=True
+    )
+    
+    return fig
+
+def predict_stock(ticker):
+    """预测流程处理"""
+    start_time = time.time()
+    
+    # 显示加载状态
+    yield "⌛ 正在获取数据并训练模型（最多30秒）...", None
+    
+    # 在后台线程中运行训练
+    result = None
+    def train_thread():
+        nonlocal result
+        result = rapid_training(ticker)
+    
+    thread = threading.Thread(target=train_thread)
+    thread.start()
+    
+    # 等待完成（最多30秒）
+    thread.join(timeout=30)
+    
+    if not result:
+        yield "❌ 训练失败或超时，请尝试其他股票代码", None
+        return
+    
+    if result['training_time'] > 30:
+        yield "⚠️ 训练超时，结果可能不准确", create_plot(result)
+        return
+    
+    # 显示结果
+    info_msg = f"✅ 训练成功（耗时{result['training_time']:.1f}秒）\n" \
+              f"最新预测：{pred.values[-1]:.2f} USD（{pred.index[-1].strftime('%Y-%m-%d')}）"
+    
+    yield info_msg, create_plot(result)
+
+with gr.Blocks() as demo:
+    gr.Markdown("# 🚀 实时股票预测系统")
+    
+    with gr.Row():
+        ticker_input = gr.Textbox(
+            label="输入股票代码",
+            placeholder="例如：AAPL（美股）, 0005.HK（港股）",
+            max_lines=1
+        )
+        submit_btn = gr.Button("立即预测", variant="primary")
+    
+    status_output = gr.Textbox(label="状态", interactive=False)
+    plot_output = gr.Plot(label="价格预测")
+    
+    submit_btn.click(
+        predict_stock,
+        inputs=ticker_input,
+        outputs=[status_output, plot_output]
+    )
+
+if __name__ == "__main__":
+    demo.launch(server_port=7860)