更新完整应用代码

app.py

@@ -3,12 +3,19 @@ import pandas as pd
 import plotly.graph_objects as go
 from prophet import Prophet
 import os
+import numpy as np
+from datetime import datetime
 
+# 设置环境变量（优化M1芯片性能）
+os.environ["OMP_NUM_THREADS"] = "1"
+os.environ["OPENBLAS_NUM_THREADS"] = "1"
+os.environ["MKL_NUM_THREADS"] = "1"
 
+# 预加载所有数据文件
 DATA_DIR = "data"
 available_tickers = {}
 
-
+# 检查data目录是否存在
 if os.path.exists(DATA_DIR):
     for filename in os.listdir(DATA_DIR):
         if filename.endswith(".csv"):
@@ -16,140 +23,341 @@ if os.path.exists(DATA_DIR):
             file_path = os.path.join(DATA_DIR, filename)
             try:
                 # 尝试不同编码读取
-                for encoding in ['utf-8', 'gbk', 'latin1']:
+                encodings = ['utf-8', 'gbk', 'latin1', 'ISO-8859-1']
+                df = None
+                
+                for encoding in encodings:
                     try:
                         df = pd.read_csv(file_path, encoding=encoding)
+                        print(f"成功使用 {encoding} 编码加载 {filename}")
                         break
-                    except:
+                    except UnicodeDecodeError:
                         continue
                 
-                # 识别日期列
-                date_cols = [col for col in df.columns if 'date' in col.lower() or 'time' in col.lower()]
-                if date_cols:
-                    df[date_cols[0]] = pd.to_datetime(df[date_cols[0]])
-                    df.set_index(date_cols[0], inplace=True)
+                if df is None:
+                    raise Exception("无法解码文件")
+                
+                # 自动检测日期列
+                date_col = None
+                for col in df.columns:
+                    if 'date' in col.lower() or 'time' in col.lower():
+                        date_col = col
+                        break
+                
+                # 如果没有找到日期列，尝试第一列
+                if date_col is None and len(df.columns) > 0:
+                    date_col = df.columns[0]
+                
+                # 转换日期列
+                if date_col:
+                    df[date_col] = pd.to_datetime(df[date_col], errors='coerce')
+                    df = df.dropna(subset=[date_col])
+                    df.set_index(date_col, inplace=True)
+                
+                # 检查必需的列
+                required_columns = ['Open', 'High', 'Low', 'Close']
+                found_columns = [col for col in required_columns if col in df.columns]
+                
+                if len(found_columns) < 3:  # 至少需要3列
+                    print(f"警告: {filename} 缺少必要列，将跳过")
+                    continue
                 
                 available_tickers[ticker_name] = {
                     "file": file_path,
                     "data": df
                 }
-                print(f"成功加载: {ticker_name}")
+                print(f"✅ 成功加载: {ticker_name} ({len(df)} 条记录)")
             except Exception as e:
-                print(f"加载失败 {filename}: {str(e)}")
+                print(f"❌ 加载失败 {filename}: {str(e)}")
 else:
-    print(f"警告: 数据目录不存在 - {DATA_DIR}")
+    print(f"⚠️ 警告: 数据目录不存在 - {DATA_DIR}")
 
 def analyze_stock(ticker):
-    if not any(available_tickers):
-        return "❌ 错误: 没有找到任何数据文件。请检查data/目录", None, None
-    
-   
-    ticker_upper = ticker.upper()
-    matched_ticker = None
-    
-  
-    if ticker_upper in available_tickers:
-        matched_ticker = ticker_upper
-    else:
-  
-        for name in available_tickers.keys():
-            if ticker_upper in name or name in ticker_upper:
-                matched_ticker = name
-                break
-    
-    if not matched_ticker:
-        return f"❌ 未找到匹配的数据: {ticker}\n可用数据: {', '.join(available_tickers.keys())}", None, None
-    
+    """分析股票/外汇数据"""
     try:
-        hist = available_tickers[matched_ticker]["data"]
+        # 检查是否有可用数据
+        if not available_tickers:
+            return "❌ 错误: 没有找到任何数据文件。请检查data/目录", None, None
+        
+        # 模糊匹配股票代码
+        ticker_upper = ticker.upper()
+        matched_ticker = None
+        
+        # 尝试精确匹配
+        if ticker_upper in available_tickers:
+            matched_ticker = ticker_upper
+        else:
+            # 尝试部分匹配
+            for name in available_tickers.keys():
+                if ticker_upper in name or name in ticker_upper:
+                    matched_ticker = name
+                    break
         
-     
+        if not matched_ticker:
+            return (f"❌ 未找到匹配的数据: {ticker}\n"
+                    f"可用数据: {', '.join(available_tickers.keys())}"), None, None
+        
+        # 获取数据
+        hist = available_tickers[matched_ticker]["data"].copy()
+        
+        # 确保必要的列存在
         required_cols = ['Open', 'High', 'Low', 'Close']
-        if not all(col in hist.columns for col in required_cols):
-            return f"❌ 数据格式错误: 缺少必要列。请确保CSV包含: {', '.join(required_cols)}", None, None
+        missing_cols = [col for col in required_cols if col not in hist.columns]
+        
+        if missing_cols:
+            # 尝试查找相似列名
+            col_mapping = {}
+            for col in missing_cols:
+                for existing_col in hist.columns:
+                    if col.lower() in existing_col.lower() or existing_col.lower() in col.lower():
+                        col_mapping[existing_col] = col
+            
+            if col_mapping:
+                hist.rename(columns=col_mapping, inplace=True)
+                missing_cols = [col for col in required_cols if col not in hist.columns]
+            
+            if missing_cols:
+                return (f"❌ 数据格式错误: 缺少必要列: {', '.join(missing_cols)}\n"
+                        f"可用列: {', '.join(hist.columns)}"), None, None
+        
+        # 创建K线图
+        fig = go.Figure(data=[go.Candlestick(
+            x=hist.index,
+            open=hist['Open'],
+            high=hist['High'],
+            low=hist['Low'],
+            close=hist['Close'],
+            name='价格'
+        )])
+        
+        # 添加移动平均线
+        if len(hist) >= 20:
+            hist['MA20'] = hist['Close'].rolling(window=20, min_periods=1).mean()
+            fig.add_trace(go.Scatter(
+                x=hist.index, 
+                y=hist['MA20'], 
+                mode='lines', 
+                name='20日均线',
+                line=dict(color='blue', width=1.5)
+            ))
+        
+        if len(hist) >= 50:
+            hist['MA50'] = hist['Close'].rolling(window=50, min_periods=1).mean()
+            fig.add_trace(go.Scatter(
+                x=hist.index, 
+                y=hist['MA50'], 
+                mode='lines', 
+                name='50日均线',
+                line=dict(color='orange', width=1.5)
+            ))
+        
+        fig.update_layout(
+            title=f"{matched_ticker} 价格走势 (本地数据)",
+            xaxis_title="日期",
+            yaxis_title="价格",
+            template="plotly_white",
+            hovermode="x unified",
+            height=500
+        )
+        
+        # 预测 - 使用Prophet (修复版本)
+        try:
+            # 准备数据
+            df = hist[['Close']].reset_index()
+            df.columns = ['ds', 'y']
+            
+            # 移除NaN值
+            df = df.dropna()
+            
+            # 确保有足够数据
+            if len(df) < 30:
+                raise ValueError("数据点不足，无法进行预测")
+            
+            # 创建并拟合模型 (修复: 移除stan_backend)
+            model = Prophet(
+                daily_seasonality=True,
+                yearly_seasonality=True,
+                interval_width=0.95,
+                uncertainty_samples=1000
+            )
+            
+            model.fit(df)
+            
+            # 创建未来数据框 (30天预测)
+            future = model.make_future_dataframe(periods=30, freq='D')
+            forecast = model.predict(future)
+            
+            # 预测图表
+            fig2 = go.Figure()
+            
+            # 历史数据
+            fig2.add_trace(go.Scatter(
+                x=df['ds'], 
+                y=df['y'], 
+                mode='lines', 
+                name='历史价格',
+                line=dict(color='blue', width=2)
+            ))
+            
+            # 预测数据
+            fig2.add_trace(go.Scatter(
+                x=forecast['ds'], 
+                y=forecast['yhat'], 
+                mode='lines', 
+                name='预测价格',
+                line=dict(color='red', width=2, dash='dash')
+            ))
+            
+            # 预测区间
+            fig2.add_trace(go.Scatter(
+                x=forecast['ds'].tolist() + forecast['ds'][::-1].tolist(),
+                y=forecast['yhat_upper'].tolist() + forecast['yhat_lower'][::-1].tolist(),
+                fill='toself',
+                fillcolor='rgba(255,0,0,0.1)',
+                line=dict(color='rgba(255,255,255,0)'),
+                name='置信区间'
+            ))
+            
+            fig2.update_layout(
+                title=f"{matched_ticker} 30天价格预测",
+                xaxis_title="日期",
+                yaxis_title="价格",
+                template="plotly_white",
+                hovermode="x unified",
+                height=500
+            )
+            
+        except Exception as e:
+            print(f"预测错误: {str(e)}")
+            fig2 = None
+            forecast_error = str(e)
+        
+        # 技术指标计算
+        if 'MA20' not in hist.columns and len(hist) >= 20:
+            hist['MA20'] = hist['Close'].rolling(20).mean()
+        
+        if 'MA50' not in hist.columns and len(hist) >= 50:
+            hist['MA50'] = hist['Close'].rolling(50).mean()
         
-    
-        fig = go.Figure(data=[go.Candlestick(x=hist.index,
-                    open=hist['Open'],
-                    high=hist['High'],
-                    low=hist['Low'],
-                    close=hist['Close'])])
-        fig.update_layout(title=f"{matched_ticker} 股票K线图 (本地数据)", xaxis_title="日期", yaxis_title="价格")
-        
- 
-        df = hist[['Close']].reset_index()
-        df.columns = ['ds', 'y']
-        model = Prophet(daily_seasonality=True, yearly_seasonality=True)
-        model.fit(df)
-        future = model.make_future_dataframe(periods=30)  # 30天预测
-        forecast = model.predict(future)
-        
-        fig2 = go.Figure()
-        fig2.add_trace(go.Scatter(x=df['ds'], y=df['y'], mode='lines', name='历史价格'))
-        fig2.add_trace(go.Scatter(x=forecast['ds'], y=forecast['yhat'], mode='lines', name='预测价格'))
-        fig2.update_layout(title=f"{matched_ticker} 30天价格预测", xaxis_title="日期", yaxis_title="价格")
-        
-     
-        hist['MA20'] = hist['Close'].rolling(20).mean()
-        hist['MA50'] = hist['Close'].rolling(50).mean()
         current_price = hist['Close'].iloc[-1]
-        ma20 = hist['MA20'].iloc[-1]
-        ma50 = hist['MA50'].iloc[-1]
-        
- 
-        if current_price > ma20 > ma50:
-            signal = "📈 强烈看涨 (黄金交叉)"
-        elif current_price > ma20:
-            signal = "📈 看涨"
-        elif current_price < ma20 < ma50:
-            signal = "📉 强烈看跌 (死亡交叉)"
-        else:
-            signal = "🔄 震荡"
+        ma20 = hist['MA20'].iloc[-1] if 'MA20' in hist.columns else None
+        ma50 = hist['MA50'].iloc[-1] if 'MA50' in hist.columns else None
+        
+        # 信号判断
+        signal = "🔍 数据不足，无法判断信号"
+        if ma20 is not None:
+            if current_price > ma20:
+                signal = "📈 看涨 (价格 > 20日均线)"
+            else:
+                signal = "📉 看跌 (价格 < 20日均线)"
+        
+        if ma20 is not None and ma50 is not None:
+            if current_price > ma20 > ma50:
+                signal = "🚀 强烈看涨 (黄金交叉)"
+            elif current_price < ma20 < ma50:
+                signal = "💣 强烈看跌 (死亡交叉)"
+        
+        # 计算回报率
+        period_return = (current_price / hist['Close'].iloc[0] - 1) * 100
         
         result_text = (
-            f"📊 {matched_ticker} 分析结果\n"
-            f"💰 当前价格: ${current_price:.2f}\n"
-            f"📈 20日均线: ${ma20:.2f}\n"
-            f"📉 50日均线: ${ma50:.2f}\n"
-            f"🎯 信号: {signal}\n"
-            f"💾 数据来源: 本地文件 ({len(hist)} 条记录)"
+            f"📊 {matched_ticker} 分析报告\n"
+            f"💰 当前价格: {current_price:.5f}\n"
+            f"📈 20日均线: {ma20:.5f}\n" if ma20 is not None else ""
+            f"📉 50日均线: {ma50:.5f}\n" if ma50 is not None else ""
+            f"🎯 交易信号: {signal}\n"
+            f"📊 总回报率: {period_return:.2f}%\n"
+            f"💾 数据记录: {len(hist)} 条\n"
+            f"🕒 更新时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}"
         )
         
         return result_text, fig, fig2
         
     except Exception as e:
-        return f"❌ 分析错误: {str(e)}", None, None
+        error_msg = f"❌ 分析错误: {str(e)}"
+        print(error_msg)
+        return error_msg, None, None
 
 def list_available_data():
+    """列出可用数据"""
     if not available_tickers:
-        return "暂无可用数据文件"
-    return "可用数据: " + ", ".join(available_tickers.keys())
+        return "⚠️ 未找到数据文件。请将CSV文件放入data/目录"
+    return "✅ 可用数据: " + ", ".join(available_tickers.keys())
+
+def get_app_version():
+    """获取应用版本信息"""
+    return f"📈 外汇/股票AI分析系统 v2.1\n" \
+           f"🕒 最后更新: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n" \
+           f"📊 可用数据集: {len(available_tickers)}"
 
-with gr.Blocks(title="股票AI分析 (本地数据版)") as demo:
-    gr.Markdown("# 📈 股票AI分析系统 (本地历史数据版)")
-    gr.Markdown("✅ 优势: 无需网络，数据稳定，适合中长期分析")
+# 创建Gradio界面
+with gr.Blocks(title="外汇AI分析系统", theme=gr.themes.Soft()) as demo:
+    gr.Markdown("# 📈 外汇/股票AI分析系统 (本地数据版)")
+    gr.Markdown("✅ 优势: 无需网络，数据稳定，隐私安全，适合中长期分析")
     
-    data_status = gr.Textbox(label="数据状态", value=list_available_data(), interactive=False)
+    # 状态信息
+    with gr.Row():
+        version_info = gr.Textbox(label="系统信息", value=get_app_version(), interactive=False)
+        data_status = gr.Textbox(label="数据状态", value=list_available_data(), interactive=False)
     
+    # 分析区域
     with gr.Row():
-        ticker_input = gr.Textbox(label="股票代码/名称", value="EURUSD")
-        analyze_btn = gr.Button("分析", variant="primary")
+        ticker_input = gr.Textbox(label="数据名称", value="EURUSD", placeholder="输入数据文件名，如: EURUSD, AAPL")
+        analyze_btn = gr.Button("分析", variant="primary", size="lg")
+    
+    # 结果区域
+    signal_output = gr.Textbox(label="分析结果", lines=6)
     
-    signal_output = gr.Textbox(label="分析结果")
     with gr.Row():
-        kchart = gr.Plot(label="K线图")
-        pred_chart = gr.Plot(label="价格预测")
+        kchart = gr.Plot(label="价格走势与技术指标")
+        pred_chart = gr.Plot(label="30天价格预测")
+    
+    # 使用指南
+    gr.Markdown("""
+    ### 📋 使用指南
+    1. **输入数据名称**：使用数据文件名（不含.csv后缀），例如 `EURUSD`
+    2. **点击"分析"**：系统将加载本地数据并生成分析报告
+    3. **查看结果**：包括技术指标、交易信号和价格预测
+    4. **数据更新**：上传新的CSV文件到data/目录，重新部署应用
     
-    gr.Markdown("### 使用指南:\n"
-               "1. 输入货币对名称（例如: EURUSD）\n"
-               "2. 系统自动从本地数据加载\n"
-               "3. 查看技术分析和30天预测\n"
-               "4. 定期更新data/目录中的CSV文件")
+    ### 🔍 数据格式要求
+    CSV文件应包含以下列（不区分大小写）：
+    - 日期列 (Date/Time)
+    - 开盘价 (Open)
+    - 最高价 (High)
+    - 最低价 (Low)
+    - 收盘价 (Close)
+    
+    ### 💡 提示
+    - 首次加载可能需要1-2分钟
+    - 数据越多，预测越准确（建议至少6个月数据）
+    - 支持多种金融产品：外汇、股票、加密货币
+    """)
+    
+    # 示例按钮
+    with gr.Row():
+        gr.Examples(
+            examples=[
+                ["EURUSD"],
+                ["AAPL"],
+                ["BTCUSD"]
+            ],
+            inputs=ticker_input,
+            label="常用数据示例",
+            examples_per_page=3
+        )
     
+    # 分析按钮点击事件
     analyze_btn.click(
         fn=analyze_stock,
         inputs=ticker_input,
         outputs=[signal_output, kchart, pred_chart]
     )
 
-demo.launch()
+# 启动应用
+if __name__ == "__main__":
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=False
+    )