# -*- coding: utf-8 -*- """Structured technical signal analysis used by technical tools.""" from dataclasses import dataclass, field from typing import Dict, List, Optional import pandas as pd @dataclass class TechnicalSignal: """Structured technical analysis result for one ticker.""" ticker: str current_price: float = 0.0 ma5: float = 0.0 ma10: float = 0.0 ma20: float = 0.0 ma50: float = 0.0 ma200: Optional[float] = None bias_ma5_pct: float = 0.0 momentum_5d_pct: float = 0.0 momentum_10d_pct: float = 0.0 momentum_20d_pct: float = 0.0 annualized_volatility_pct: float = 0.0 rsi14: float = 50.0 macd: float = 0.0 macd_signal: float = 0.0 bollinger_upper: float = 0.0 bollinger_mid: float = 0.0 bollinger_lower: float = 0.0 trend: str = "NEUTRAL" mean_reversion_signal: str = "NEUTRAL" risk_level: str = "MODERATE RISK" notes: List[str] = field(default_factory=list) def to_summary(self) -> Dict[str, object]: """Compact dict for logs/tests.""" return { "ticker": self.ticker, "trend": self.trend, "mean_reversion_signal": self.mean_reversion_signal, "risk_level": self.risk_level, "current_price": self.current_price, "rsi14": self.rsi14, "annualized_volatility_pct": self.annualized_volatility_pct, } class StockTechnicalAnalyzer: """Lightweight technical analyzer adapted for 大时代 tools.""" def analyze(self, ticker: str, df: pd.DataFrame) -> TechnicalSignal: """Analyze one ticker from OHLC price history.""" result = TechnicalSignal(ticker=ticker) if df is None or df.empty or len(df) < 5: result.notes.append("Insufficient price data") return result frame = df.sort_values("time").reset_index(drop=True).copy() frame["close"] = pd.to_numeric(frame["close"], errors="coerce") frame["returns"] = frame["close"].pct_change() for window in (5, 10, 20, 50, 200): frame[f"MA_{window}"] = frame["close"].rolling(window).mean() frame["EMA_12"] = frame["close"].ewm(span=12, adjust=False).mean() frame["EMA_26"] = frame["close"].ewm(span=26, adjust=False).mean() frame["MACD"] = frame["EMA_12"] - frame["EMA_26"] frame["MACD_SIGNAL"] = ( frame["MACD"].ewm(span=9, adjust=False).mean() ) delta = frame["close"].diff() gain = delta.where(delta > 0, 0.0) loss = -delta.where(delta < 0, 0.0) avg_gain = gain.rolling(14).mean() avg_loss = loss.rolling(14).mean() rs = avg_gain / avg_loss.replace(0, pd.NA) frame["RSI_14"] = 100 - (100 / (1 + rs)) frame["BB_MID"] = frame["close"].rolling(20).mean() frame["BB_STD"] = frame["close"].rolling(20).std() frame["BB_UPPER"] = frame["BB_MID"] + 2 * frame["BB_STD"] frame["BB_LOWER"] = frame["BB_MID"] - 2 * frame["BB_STD"] latest = frame.iloc[-1] result.current_price = _safe_number(latest["close"]) result.ma5 = _safe_number(latest["MA_5"]) result.ma10 = _safe_number(latest["MA_10"]) result.ma20 = _safe_number(latest["MA_20"]) result.ma50 = _safe_number(latest["MA_50"]) result.ma200 = _safe_optional(latest["MA_200"]) result.bias_ma5_pct = _percent_gap(result.current_price, result.ma5) result.momentum_5d_pct = _lookback_return(frame["close"], 5) result.momentum_10d_pct = _lookback_return(frame["close"], 10) result.momentum_20d_pct = _lookback_return(frame["close"], 20) result.annualized_volatility_pct = _safe_number( frame["returns"].tail(20).std() * (252**0.5) * 100, ) result.rsi14 = _safe_number(latest["RSI_14"], default=50.0) result.macd = _safe_number(latest["MACD"]) result.macd_signal = _safe_number(latest["MACD_SIGNAL"]) result.bollinger_mid = _safe_number(latest["BB_MID"]) result.bollinger_upper = _safe_number(latest["BB_UPPER"]) result.bollinger_lower = _safe_number(latest["BB_LOWER"]) result.trend = _classify_trend(result) result.mean_reversion_signal = _classify_mean_reversion(result) result.risk_level = _classify_risk(result.annualized_volatility_pct) result.notes = _build_notes(result) return result def _safe_number(value, default: float = 0.0) -> float: try: if pd.isna(value): return default return float(value) except (TypeError, ValueError): return default def _safe_optional(value) -> Optional[float]: try: if pd.isna(value): return None return float(value) except (TypeError, ValueError): return None def _lookback_return(series: pd.Series, lookback: int) -> float: if len(series) <= lookback: return 0.0 base = _safe_number(series.iloc[-lookback - 1]) latest = _safe_number(series.iloc[-1]) if base <= 0: return 0.0 return ((latest / base) - 1) * 100 def _percent_gap(value: float, anchor: float) -> float: if anchor <= 0: return 0.0 return ((value - anchor) / anchor) * 100 def _classify_trend(result: TechnicalSignal) -> str: bullish_stack = ( result.current_price >= result.ma5 >= result.ma10 >= result.ma20 > 0 ) if bullish_stack and result.macd >= result.macd_signal: return "STRONG BULLISH" if bullish_stack: return "BULLISH" if result.current_price < result.ma20 and result.macd < result.macd_signal: return "BEARISH" return "NEUTRAL" def _classify_mean_reversion(result: TechnicalSignal) -> str: if result.rsi14 <= 30 or ( result.bollinger_lower > 0 and result.current_price <= result.bollinger_lower ): return "OVERSOLD" if result.rsi14 >= 70 or ( result.bollinger_upper > 0 and result.current_price >= result.bollinger_upper ): return "OVERBOUGHT" return "NEUTRAL" def _classify_risk(volatility_pct: float) -> str: if volatility_pct > 50: return "HIGH RISK" if volatility_pct > 25: return "MODERATE RISK" return "LOW RISK" def _build_notes(result: TechnicalSignal) -> List[str]: notes = [] if abs(result.bias_ma5_pct) > 5: notes.append("Price extended from MA5") if result.macd > result.macd_signal: notes.append("MACD supports upside momentum") if result.mean_reversion_signal == "OVERSOLD": notes.append("Potential rebound setup") if result.mean_reversion_signal == "OVERBOUGHT": notes.append("Potential pullback setup") return notes