backend/app/data_manager.py

# backend/app/data_manager.py

import yfinance as yf
import pandas as pd
import numpy as np
import datetime
from app.data_cache import get_cached, save_cache

# 默认时间间隔对应的回测历史区间
INTERVAL_TO_PERIOD = {
    "1m": "5d",
    "5m": "1mo",
    "15m": "1mo",
    "30m": "1mo",
    "1h": "3mo",
    "1d": "1y"
}

# 缓存公司基本信息，避免频繁网络请求导致缓慢
COMPANY_INFO_CACHE = {}

def get_company_info(ticker):
    """
    获取公司名、板块、行业、市值和业务描述，带有缓存与默认兜底，保证极速响应。
    """
    ticker = ticker.upper()
    if ticker in COMPANY_INFO_CACHE:
        return COMPANY_INFO_CACHE[ticker]
        
    # 常用股票静态数据库，优先返回以保证零延迟
    static_db = {
        "TSLA": {
            "name": "Tesla, Inc.",
            "sector": "Consumer Cyclical (消费周期性)",
            "industry": "Auto Manufacturers (汽车制造商)",
            "market_cap": 820000000000,
            "description": "Tesla Inc. 是一家设计、开发、制造和销售电动汽车、能源生成和存储系统的美国跨国公司。它是全球最受关注的高波动率日内交易标的。"
        },
        "NVDA": {
            "name": "NVIDIA Corporation",
            "sector": "Technology (科技)",
            "industry": "Semiconductors (半导体)",
            "market_cap": 3150000000000,
            "description": "NVIDIA Corporation 是一家设计图形处理器（GPU）的半导体跨国科技公司，在人工智能芯片、数据中心和高性能计算领域处于绝对垄断地位。"
        },
        "AAPL": {
            "name": "Apple Inc.",
            "sector": "Technology (科技)",
            "industry": "Consumer Electronics (消费电子)",
            "market_cap": 3320000000000,
            "description": "Apple Inc. 是全球最具价值的电子科技公司，主营 iPhone、Mac、iPad 等消费终端设备以及各种云端订阅软件服务，现金流充裕，波动相对稳健。"
        },
        "MSFT": {
            "name": "Microsoft Corporation",
            "sector": "Technology (科技)",
            "industry": "Software—Infrastructure (基础软件)",
            "market_cap": 3250000000000,
            "description": "Microsoft Corporation 是全球软件与云服务的龙头企业。旗下拥有 Windows 系统、Azure 云平台、Office 软件，并通过 OpenAI 领跑生成式 AI 时代。"
        },
        "AMD": {
            "name": "Advanced Micro Devices, Inc.",
            "sector": "Technology (科技)",
            "industry": "Semiconductors (半导体)",
            "market_cap": 260000000000,
            "description": "Advanced Micro Devices, Inc. 是一家全球半导体公司，主营微处理器（CPU）、显卡（GPU）以及游戏主机定制芯片，与英特尔和英伟达呈竞争关系。"
        }
    }
    
    if ticker in static_db:
        COMPANY_INFO_CACHE[ticker] = static_db[ticker]
        return static_db[ticker]
        
    try:
        stock = yf.Ticker(ticker)
        info = stock.info
        name = info.get("longName", info.get("shortName", ticker))
        sector = info.get("sector", "General Sector")
        industry = info.get("industry", "General Industry")
        market_cap = info.get("marketCap", 0)
        description = info.get("longBusinessSummary", "No details available.")
        
        data = {
            "name": name,
            "sector": sector,
            "industry": industry,
            "market_cap": market_cap,
            "description": description
        }
        COMPANY_INFO_CACHE[ticker] = data
        return data
    except Exception as e:
        # 如果 yfinance 接口请求出错/被限制，使用兜底值
        fallback = {
            "name": f"{ticker} Corporation",
            "sector": "General Sector (常规板块)",
            "industry": "General Industry (常规行业)",
            "market_cap": 0,
            "description": f"未能获取到 {ticker} 的网络实时介绍，已自动生成默认档案。该标的目前可参与量化行情回测。"
        }
        COMPANY_INFO_CACHE[ticker] = fallback
        return fallback

def calculate_rsi(series, period=14):
    """
    计算 RSI (相对强弱指标)
    """
    delta = series.diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
    
    rs = gain / loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

def calculate_atr(df, period=14):
    """
    计算 ATR (真实波幅)
    """
    high_low = df['High'] - df['Low']
    high_close = np.abs(df['High'] - df['Close'].shift())
    low_close = np.abs(df['Low'] - df['Close'].shift())
    
    ranges = pd.concat([high_low, high_close, low_close], axis=1)
    true_range = np.max(ranges, axis=1)
    return true_range.rolling(window=period).mean()

def calculate_adx(df, period=14):
    """
    计算 ADX (平均趋向指数)，衡量趋势强度
    """
    df = df.copy()
    high = df['High']
    low = df['Low']
    close = df['Close']
    
    # True Range
    tr1 = high - low
    tr2 = (high - close.shift(1)).abs()
    tr3 = (low - close.shift(1)).abs()
    tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
    
    # Directional Movement
    up_move = high.diff(1)
    down_move = -low.diff(1)
    
    plus_dm = np.where((up_move > down_move) & (up_move > 0), up_move, 0.0)
    minus_dm = np.where((down_move > up_move) & (down_move > 0), down_move, 0.0)
    
    # Smoothed Wilder's MA
    tr_smooth = pd.Series(tr).ewm(alpha=1/period, adjust=False).mean()
    plus_dm_smooth = pd.Series(plus_dm, index=df.index).ewm(alpha=1/period, adjust=False).mean()
    minus_dm_smooth = pd.Series(minus_dm, index=df.index).ewm(alpha=1/period, adjust=False).mean()
    
    plus_di = 100 * (plus_dm_smooth / np.maximum(tr_smooth, 1e-8))
    minus_di = 100 * (minus_dm_smooth / np.maximum(tr_smooth, 1e-8))
    
    dx = 100 * (plus_di - minus_di).abs() / np.maximum(plus_di + minus_di, 1e-8)
    adx = dx.ewm(alpha=1/period, adjust=False).mean()
    
    return plus_di, minus_di, adx

def get_yesterday_levels(ticker):
    """
    获取昨日的最高价(PDH)、最低价(PDL)、收盘价(PDC)
    使用日线级别数据，确保绝对准确
    """
    try:
        stock = yf.Ticker(ticker)
        df_daily = stock.history(period="5d")
        if len(df_daily) >= 2:
            yesterday = df_daily.iloc[-2]  # -1 是今天，-2 是昨天
            return {
                "PDH": float(yesterday['High']),
                "PDL": float(yesterday['Low']),
                "PDC": float(yesterday['Close'])
            }
    except Exception as e:
        print(f"获取昨日关键位置失败 ({ticker}): {str(e)}")
    
    return {"PDH": 0.0, "PDL": 0.0, "PDC": 0.0}

def compute_candle_features(df):
    """
    计算K线形态特征向量：实体比例、上影线比例、下影线比例、跳空比例
    """
    df = df.copy()
    diff = df['High'] - df['Low']
    denom = np.where(diff == 0, 1e-8, diff)
    
    df['body_ratio'] = np.abs(df['Close'] - df['Open']) / denom
    df['upper_shadow_ratio'] = (df['High'] - np.maximum(df['Close'], df['Open'])) / denom
    df['lower_shadow_ratio'] = (np.minimum(df['Close'], df['Open']) - df['Low']) / denom
    df['gap_flag'] = (df['Open'] - df['Close'].shift(1)) / (df['Close'].shift(1) + 1e-8)
    
    df['body_ratio'] = df['body_ratio'].fillna(0.0)
    df['upper_shadow_ratio'] = df['upper_shadow_ratio'].fillna(0.0)
    df['lower_shadow_ratio'] = df['lower_shadow_ratio'].fillna(0.0)
    df['gap_flag'] = df['gap_flag'].fillna(0.0)
    return df

def fetch_and_prepare_data(ticker, period=None, interval="1m"):
    """
    获取股票行情并计算量化指标，支持多时间周期。
    """
    ticker = ticker.upper()
    if period is None:
        period = INTERVAL_TO_PERIOD.get(interval, "5d")
        
    # 优先尝试从本地缓存读取
    cached_df = get_cached(ticker, period, interval)
    if cached_df is not None:
        return cached_df
        
    # 抓取包含盘前盘后的 K线数据
    stock = yf.Ticker(ticker)
    
    # 只有分钟级别 (1m, 5m, 15m, 30m, 1h) 支持盘前盘后 prepost
    is_intraday = interval in ["1m", "5m", "15m", "30m", "1h"]
    
    df = stock.history(period=period, interval=interval, prepost=is_intraday)
    
    if df.empty:
        raise ValueError(f"未能获取到 {ticker} 的 {interval} 行情数据。")
        
    # 确保时间戳已转换为本地时区 (美东时间)
    if df.index.tz is None:
        df = df.tz_localize('UTC').tz_convert('America/New_York')
    else:
        df = df.tz_convert('America/New_York')
        
    df['Date'] = df.index.date
    
    # 计算 VWAP (日内均线按交易日独立累计，如果是日线级别，VWAP 退化为典型价)
    df['Typical_Price'] = (df['High'] + df['Low'] + df['Close']) / 3
    df['TP_Volume'] = df['Typical_Price'] * df['Volume']
    
    if is_intraday:
        df['Cum_TP_Vol'] = df.groupby('Date')['TP_Volume'].cumsum()
        df['Cum_Vol'] = df.groupby('Date')['Volume'].cumsum()
        # 避免除以 0
        df['Cum_Vol'] = df['Cum_Vol'].replace(0, 1)
        df['VWAP'] = df['Cum_TP_Vol'] / df['Cum_Vol']
    else:
        df['VWAP'] = df['Typical_Price']
        
    # 计算 9 / 21 / 50 EMA 和 RSI
    df['EMA_9'] = df['Close'].ewm(span=9, adjust=False).mean()
    df['EMA_21'] = df['Close'].ewm(span=21, adjust=False).mean()
    df['EMA_50'] = df['Close'].ewm(span=50, adjust=False).mean()
    df['RSI'] = calculate_rsi(df['Close'], period=14)
    df['ATR'] = calculate_atr(df, period=14)
    
    # 补充指标：ADX, MACD, Donchian, OBV, ROC, RVOL
    df['Plus_DI'], df['Minus_DI'], df['ADX'] = calculate_adx(df, period=14)
    
    # MACD
    ema12 = df['Close'].ewm(span=12, adjust=False).mean()
    ema26 = df['Close'].ewm(span=26, adjust=False).mean()
    df['MACD'] = ema12 - ema26
    df['MACD_Signal'] = df['MACD'].ewm(span=9, adjust=False).mean()
    df['MACD_Hist'] = df['MACD'] - df['MACD_Signal']
    
    # 唐奇安通道 (应用 .shift(1) 排除当前K线自身，彻底杜绝未来函数前瞻偏差)
    df['Donchian_High'] = df['High'].rolling(window=20).max().shift(1)
    df['Donchian_Low'] = df['Low'].rolling(window=20).min().shift(1)
    df['Donchian_High_55'] = df['High'].rolling(window=55).max().shift(1)
    df['Donchian_Low_55'] = df['Low'].rolling(window=55).min().shift(1)
    
    # 能量潮指标 (OBV)
    df['OBV'] = (np.sign(df['Close'].diff()).fillna(0.0) * df['Volume']).cumsum()
    
    # 动量 ROC
    df['ROC'] = df['Close'].pct_change(periods=20) * 100
    
    # 相对成交量 RVOL
    df['RVOL'] = df['Volume'] / np.maximum(df['Volume'].rolling(window=20).mean(), 1e-8)
    
    # 计算 TTM Squeeze 挤压状态
    df['BB_Basis'] = df['Close'].rolling(window=20).mean()
    df['BB_Std'] = df['Close'].rolling(window=20).std()
    df['BB_Upper'] = df['BB_Basis'] + (2 * df['BB_Std'])
    df['BB_Lower'] = df['BB_Basis'] - (2 * df['BB_Std'])
    
    # 肯特纳通道 (标准：基于 20 EMA 和 2.0 ATR)
    df['KC_Basis'] = df['Close'].ewm(span=20, adjust=False).mean()
    df['KC_Upper'] = df['KC_Basis'] + (2.0 * df['ATR'])
    df['KC_Lower'] = df['KC_Basis'] - (2.0 * df['ATR'])
    
    df['Squeeze_On'] = (df['BB_Upper'] < df['KC_Upper']) & (df['BB_Lower'] > df['KC_Lower'])
    
    # 状态路由判定 (Regime Classification)
    df['ATR_Ratio'] = df['ATR'] / df['Close']
    high_vol_th = df['ATR_Ratio'].rolling(252, min_periods=20).quantile(0.90).fillna(0.05)
    
    regimes = np.array(["range_bound"] * len(df), dtype=object)
    
    is_high_vol = df['ATR_Ratio'] > high_vol_th
    is_trend_up = (df['ADX'] > 20) & (df['EMA_9'] > df['EMA_21']) & (df['EMA_21'] > df['EMA_50'])
    is_trend_down = (df['ADX'] > 20) & (df['EMA_9'] < df['EMA_21']) & (df['EMA_21'] < df['EMA_50'])
    
    # 优先级别：高波动 -> 下跌趋势 -> 上涨趋势 -> 震荡
    regimes[is_trend_down] = "trend_down"
    regimes[is_trend_up] = "trend_up"
    regimes[is_high_vol] = "high_volatility"
    
    df['Regime'] = regimes
    
    # 盘前与昨日关键位
    if is_intraday:
        df['Time'] = df.index.time
        market_open = datetime.time(9, 30)
        market_close = datetime.time(16, 0)
        
        df['Is_Regular_Hours'] = df['Time'].apply(lambda t: market_open <= t <= market_close)
        df['Is_Pre_Market'] = df['Time'].apply(lambda t: datetime.time(4, 0) <= t < market_open)
        
        # 每日计算 PMH / PML 盘前最值
        pm_data = df[df['Is_Pre_Market']]
        if not pm_data.empty:
            pmh_dict = pm_data.groupby('Date')['High'].max().to_dict()
            pml_dict = pm_data.groupby('Date')['Low'].min().to_dict()
            df['PMH'] = df['Date'].map(pmh_dict).fillna(0.0)
            df['PML'] = df['Date'].map(pml_dict).fillna(0.0)
        else:
            df['PMH'] = 0.0
            df['PML'] = 0.0

        # 每日计算 ORB (开盘前5分钟 9:30-9:35 区间高低点)
        orb_data = df[(df['Time'] >= datetime.time(9, 30)) & (df['Time'] <= datetime.time(9, 35))]
        if not orb_data.empty:
            orb_h_dict = orb_data.groupby('Date')['High'].max().to_dict()
            orb_l_dict = orb_data.groupby('Date')['Low'].min().to_dict()
            df['ORB_High'] = df['Date'].map(orb_h_dict).fillna(0.0)
            df['ORB_Low'] = df['Date'].map(orb_l_dict).fillna(0.0)
        else:
            df['ORB_High'] = 0.0
            df['ORB_Low'] = 0.0
            
        regular_hours_df = df[df['Is_Regular_Hours']].copy()
        if len(regular_hours_df) < 2:
            # 当处在非常规交易时段或盘前导致常规时段K线不足时，自动回退取全部可用K线(包含盘前/盘后)
            regular_hours_df = df.copy()
    else:
        # 日线级别数据全部视为常规时段
        regular_hours_df = df.copy()
        regular_hours_df['Is_Regular_Hours'] = True
        regular_hours_df['Is_Pre_Market'] = False
        regular_hours_df['PMH'] = 0.0
        regular_hours_df['PML'] = 0.0
        regular_hours_df['ORB_High'] = 0.0
        regular_hours_df['ORB_Low'] = 0.0
        
    # 获取昨日关键位
    yesterday_levels = get_yesterday_levels(ticker)
    regular_hours_df['PDH'] = yesterday_levels['PDH']
    regular_hours_df['PDL'] = yesterday_levels['PDL']
    regular_hours_df['PDC'] = yesterday_levels['PDC']
    
    # 清除临时列
    regular_hours_df.drop(columns=['Typical_Price', 'TP_Volume', 'Cum_TP_Vol', 'Cum_Vol'], inplace=True, errors='ignore')
    
    # 计算 K线形态特征向量
    regular_hours_df = compute_candle_features(regular_hours_df)
    
    # 填充缺失值，避免初期的 NaN 导致崩溃
    regular_hours_df.ffill(inplace=True)
    regular_hours_df.bfill(inplace=True)
    
    # 保存数据到 Parquet 本地缓存
    save_cache(ticker, period, interval, regular_hours_df)
    
    return regular_hours_df

BATCH_QUOTES_CACHE = {}
BATCH_QUOTES_TIMESTAMP = None

def get_batch_quotes(tickers: list):
    """
    极速批量获取自选股实时/最新交易日价格、前收价、当日涨跌额、涨跌幅(%)、最高价、最低价与成交量。
    带有 15 秒内存缓存与回退兜底，防 Rate Limit 且响应极快。
    """
    global BATCH_QUOTES_CACHE, BATCH_QUOTES_TIMESTAMP
    
    now = datetime.datetime.now()
    clean_tickers = [str(t).strip().upper() for t in tickers if str(t).strip()]
    if not clean_tickers:
        return {}

    # 15s 内存缓存有效性校验
    if BATCH_QUOTES_TIMESTAMP and (now - BATCH_QUOTES_TIMESTAMP).total_seconds() < 15:
        cached_res = {t: BATCH_QUOTES_CACHE[t] for t in clean_tickers if t in BATCH_QUOTES_CACHE}
        if len(cached_res) == len(clean_tickers):
            return cached_res

    results = {}
    
    # 优先使用 Alpaca Data API 抓取 100% 准确的实盘 Snapshot 报价（含前收盘价与当日真实涨跌幅）
    try:
        from app.config import ALPACA_API_KEY, ALPACA_SECRET_KEY
        if ALPACA_API_KEY and ALPACA_SECRET_KEY:
            from alpaca.data.historical import StockHistoricalDataClient
            from alpaca.data.requests import StockSnapshotRequest
            
            data_client = StockHistoricalDataClient(ALPACA_API_KEY, ALPACA_SECRET_KEY)
            snapshots = data_client.get_stock_snapshot(StockSnapshotRequest(symbol_or_symbols=clean_tickers))
            
            for ticker, snap in snapshots.items():
                if not snap:
                    continue
                cur_p = float(snap.latest_trade.price) if snap.latest_trade else float(snap.daily_bar.close if snap.daily_bar else 0.0)
                prev_p = float(snap.previous_daily_bar.close) if snap.previous_daily_bar else cur_p
                
                if cur_p > 0 and prev_p > 0:
                    chg = cur_p - prev_p
                    pct = (chg / prev_p * 100.0)
                    day_high = float(snap.daily_bar.high) if snap.daily_bar else cur_p
                    day_low = float(snap.daily_bar.low) if snap.daily_bar else cur_p
                    volume = int(snap.daily_bar.volume) if snap.daily_bar else 0
                    
                    quote_data = {
                        "ticker": ticker,
                        "price": round(cur_p, 2),
                        "prev_close": round(prev_p, 2),
                        "change": round(chg, 2),
                        "change_percent": round(pct, 2),
                        "high": round(day_high, 2),
                        "low": round(day_low, 2),
                        "volume": volume,
                        "timestamp": now.isoformat()
                    }
                    results[ticker] = quote_data
                    BATCH_QUOTES_CACHE[ticker] = quote_data
    except Exception as e:
        print(f"Alpaca snapshot quotes warning: {e}")

    # 若有未覆盖到的 ticker，使用 yfinance history 准确计算 prev_close 补救
    missing = [t for t in clean_tickers if t not in results]
    if missing:
        for ticker in missing:
            if ticker in BATCH_QUOTES_CACHE and (now - BATCH_QUOTES_TIMESTAMP).total_seconds() < 15 if BATCH_QUOTES_TIMESTAMP else False:
                results[ticker] = BATCH_QUOTES_CACHE[ticker]
                continue
            try:
                stk = yf.Ticker(ticker)
                hist = stk.history(period="5d")
                if hist is not None and not hist.empty:
                    latest_close = float(hist['Close'].iloc[-1])
                    prev_close = float(hist['Close'].iloc[-2]) if len(hist) >= 2 else latest_close
                    change = latest_close - prev_close
                    change_pct = (change / prev_close * 100.0) if prev_close != 0 else 0.0
                    
                    quote_data = {
                        "ticker": ticker,
                        "price": round(latest_close, 2),
                        "prev_close": round(prev_close, 2),
                        "change": round(change, 2),
                        "change_percent": round(change_pct, 2),
                        "high": round(float(hist['High'].iloc[-1]), 2),
                        "low": round(float(hist['Low'].iloc[-1]), 2),
                        "volume": int(hist['Volume'].iloc[-1]),
                        "timestamp": now.isoformat()
                    }
                    results[ticker] = quote_data
                    BATCH_QUOTES_CACHE[ticker] = quote_data
            except Exception:
                pass

    # 再次补漏 fallback 逻辑，保证前端100%有数据返回
    for ticker in clean_tickers:
        if ticker not in results:
            if ticker in BATCH_QUOTES_CACHE:
                results[ticker] = BATCH_QUOTES_CACHE[ticker]
            else:
                fallback_data = {
                    "ticker": ticker,
                    "price": 100.0,
                    "prev_close": 100.0,
                    "change": 0.0,
                    "change_percent": 0.0,
                    "high": 100.0,
                    "low": 100.0,
                    "volume": 0,
                    "timestamp": now.isoformat()
                }
                results[ticker] = fallback_data

    BATCH_QUOTES_TIMESTAMP = now
    return results

if __name__ == "__main__":
    print("测试多周期获取数据...")
    for iv in ["5m", "1d"]:
        data = fetch_and_prepare_data("AAPL", interval=iv)
        print(f"周期 {iv}: 获取到 {len(data)} 行数据")
        info = get_company_info("AAPL")
        print(f"公司名: {info['name']}, 行业: {info['industry']}")