single_stock.py
"""
单票量化策略模块
该模块仅负责单票分析判断，不做全局股票池遍历、数据批量获取
"""
import logging
from typing import Dict, List, Any
import pandas as pd
# 单票量化策略模块日志
logger = logging.getLogger("AIQuant.single_stock")
logger.setLevel(logging.INFO)
# 导入策略配置
from config.strategy_config import SINGLE_STOCK_CONFIG
class StockAnalysisResult:
"""
股票分析结果类
标准化输出结构
"""
def __init__(self,
symbol: str,
score: float,
reasons: List[str],
rating: str,
signal: str,
metrics: Dict[str, float]):
"""
初始化股票分析结果
Args:
symbol: 股票代码
score: 综合评分（0-100）
reasons: 评分理由（分点字符串）
rating: 标的等级（S/A/B/C）
signal: 交易信号（买入/持有/卖出）
metrics: 关键指标数值（字典）
"""
self.symbol = symbol
self.score = score
self.reasons = reasons
self.rating = rating
self.signal = signal
self.metrics = metrics
def to_dict(self) -> Dict[str, Any]:
"""
转换为字典格式
Returns:
分析结果字典
"""
return {
"股票代码": self.symbol,
"综合评分": self.score,
"评分理由": self.reasons,
"标的等级": self.rating,
"交易信号": self.signal,
"关键指标": self.metrics
}
def __str__(self) -> str:
"""
字符串表示
Returns:
字符串表示
"""
return (f"股票代码: {self.symbol}, 综合评分: {self.score:.2f}, "
f"标的等级: {self.rating}, 交易信号: {self.signal}, "
f"关键指标: {self.metrics}")
class SingleStockAnalyzer:
"""
单票量化分析器
仅负责单票分析判断，不做全局股票池遍历、数据批量获取
"""
def __init__(self, config: Dict[str, Any] = None):
"""
初始化单票量化分析器
Args:
config: 分析器配置参数
"""
self.config = config or {}
self.logger = logger
def analyze(
self,
symbol: str,
kline_data: pd.DataFrame
) -> StockAnalysisResult:
"""
分析单只股票
Args:
symbol: 股票代码（str）
kline_data: 单票完整 K 线数据（DataFrame，含 OHLCV+成交量+均线等基础指标）
Returns:
股票分析结果（StockAnalysisResult）
"""
self.logger.info(f"开始分析股票: {symbol}")
self.logger.debug(f"K 线数据形状: {kline_data.shape}")
# 初始化分析结果
score = 0.0
reasons = []
rating = "C"
signal = "卖出"
metrics = {}
try:
# 1. 数据质量检查
if self._check_data_quality(kline_data):
reasons.append("数据质量良好")
score += 10
else:
reasons.append("数据质量较差")
score += 2
# 2. 趋势分析
trend_score, trend_reasons = self._analyze_trend(kline_data)
score += trend_score
reasons.extend(trend_reasons)
# 3. 技术指标分析
tech_score, tech_reasons, tech_metrics = \
self._analyze_technical_indicators(kline_data)
score += tech_score
reasons.extend(tech_reasons)
metrics.update(tech_metrics)
# 4. 成交量分析
volume_score, volume_reasons, volume_metrics = \
self._analyze_volume(kline_data)
score += volume_score
reasons.extend(volume_reasons)
metrics.update(volume_metrics)
# 5. 风险评估
risk_score, risk_reasons, risk_metrics = self._analyze_risk(
kline_data)
score += risk_score
reasons.extend(risk_reasons)
metrics.update(risk_metrics)
# 6. 计算综合评分和等级
score = min(max(score, 0), 100)
rating = self._calculate_rating(score)
# 7. 确定交易信号
signal = self._determine_trading_signal(score, kline_data)
self.logger.info(
f"股票分析完成: {symbol} -> 评分: {score:.2f}, "
f"等级: {rating}, 信号: {signal}")
except Exception as e:
self.logger.error(f"分析股票 {symbol} 时出错: {e}")
reasons.append(f"分析过程中出现错误: {str(e)}")
return StockAnalysisResult(
symbol=symbol,
score=score,
reasons=reasons,
rating=rating,
signal=signal,
metrics=metrics
)
def _check_data_quality(self, kline_data: pd.DataFrame) -> bool:
"""
检查数据质量
Args:
kline_data: K 线数据
Returns:
数据质量是否良好的布尔值
"""
# 检查数据完整性
if kline_data.empty:
return False
# 检查必要列是否存在
required_columns = ["open", "high", "low", "close", "volume"]
for col in required_columns:
if col not in kline_data.columns:
return False
# 检查数据点数
if len(kline_data) < SINGLE_STOCK_CONFIG[
"data_quality"]["min_data_points"]:
return False
# 检查是否有缺失值
missing_ratio = kline_data.isnull().sum().sum(
) / (kline_data.shape[0] * kline_data.shape[1])
if missing_ratio > SINGLE_STOCK_CONFIG[
"data_quality"]["max_missing_ratio"]:
return False
return True
def _analyze_trend(self, kline_data: pd.DataFrame) -> (float, List[str]):
"""
分析趋势
Args:
kline_data: K 线数据
Returns:
(趋势得分, 趋势分析理由)
"""
score = 0.0
reasons = []
# 检查长期趋势
long_term_window = SINGLE_STOCK_CONFIG["trend"]["long_term_window"]
if len(kline_data) >= long_term_window:
kline_data[f"ma{long_term_window}"] = kline_data["close"].rolling(
window=long_term_window).mean()
current_price = kline_data["close"].iloc[-1]
ma_long = kline_data[f"ma{long_term_window}"].iloc[-1]
if current_price > ma_long:
score += 20
reasons.append(f"价格在 {long_term_window} 日均线上方，处于上升趋势")
else:
score += 5
reasons.append(f"价格在 {long_term_window} 日均线下方，处于下降趋势")
# 检查中期趋势
short_term_window = SINGLE_STOCK_CONFIG["trend"]["short_term_window"]
if len(kline_data) >= short_term_window:
kline_data[f"ma{short_term_window}"] = kline_data["close"].rolling(
window=short_term_window).mean()
current_price = kline_data["close"].iloc[-1]
ma_short = kline_data[f"ma{short_term_window}"].iloc[-1]
if current_price > ma_short:
score += 15
reasons.append(f"价格在 {short_term_window} 日均线上方，中期趋势向上")
else:
score += 5
reasons.append(f"价格在 {short_term_window} 日均线下方，中期趋势向下")
# 应用权重
score *= SINGLE_STOCK_CONFIG["trend"]["trend_score_weight"]
return score, reasons
def _analyze_technical_indicators(self,
kline_data: pd.DataFrame) -> (
float, List[str], Dict[str, float]
):
"""
分析技术指标
Args:
kline_data: K 线数据
Returns:
(技术指标得分, 技术指标分析理由, 技术指标数值)
"""
score = 0.0
reasons = []
metrics = {}
# 计算 RSI
rsi_window = SINGLE_STOCK_CONFIG["technical"]["rsi_window"]
if len(kline_data) >= rsi_window:
delta = kline_data["close"].diff()
gain = (delta.where(delta > 0, 0)).rolling(
window=rsi_window).mean()
loss = (-delta.where(delta < 0, 0)
).rolling(window=rsi_window).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
metrics["rsi"] = rsi.iloc[-1]
rsi_normal_range = SINGLE_STOCK_CONFIG["technical"][
"rsi_normal_range"]
rsi_overbought = SINGLE_STOCK_CONFIG["technical"]["rsi_overbought"]
rsi_oversold = SINGLE_STOCK_CONFIG["technical"]["rsi_oversold"]
if rsi_normal_range[0] < rsi.iloc[-1] < rsi_normal_range[1]:
score += 15
reasons.append(f"RSI 值为 {rsi.iloc[-1]:.2f}，处于正常区间")
elif rsi.iloc[-1] <= rsi_oversold:
score += 5
reasons.append(f"RSI 值为 {rsi.iloc[-1]:.2f}，处于超卖区")
elif rsi.iloc[-1] >= rsi_overbought:
score += 5
reasons.append(f"RSI 值为 {rsi.iloc[-1]:.2f}，处于超买区")
# 计算 MACD
macd_fast = SINGLE_STOCK_CONFIG["technical"]["macd_fast"]
macd_slow = SINGLE_STOCK_CONFIG["technical"]["macd_slow"]
macd_signal = SINGLE_STOCK_CONFIG["technical"]["macd_signal"]
if len(kline_data) >= macd_slow:
kline_data[f"ema{macd_fast}"] = kline_data["close"].ewm(
span=macd_fast, adjust=False).mean()
kline_data[f"ema{macd_slow}"] = kline_data["close"].ewm(
span=macd_slow, adjust=False).mean()
kline_data["macd"] = kline_data[f"ema{macd_fast}"] - \
kline_data[f"ema{macd_slow}"]
kline_data["macd_signal"] = kline_data["macd"].ewm(
span=macd_signal, adjust=False).mean()
metrics["macd"] = kline_data["macd"].iloc[-1]
metrics["macd_signal"] = kline_data["macd_signal"].iloc[-1]
if kline_data["macd"].iloc[-1] > \
kline_data["macd_signal"].iloc[-1]:
score += 15
reasons.append("MACD 金叉，买入信号")
else:
score += 5
reasons.append("MACD 死叉，卖出信号")
# 应用权重
score *= SINGLE_STOCK_CONFIG["technical"]["technical_score_weight"]
return score, reasons, metrics
def _analyze_volume(self,
kline_data: pd.DataFrame) -> (float,
List[str],
Dict[str, float]):
"""
分析成交量
Args:
kline_data: K 线数据
Returns:
(成交量得分, 成交量分析理由, 成交量指标数值)
"""
score = 0.0
reasons = []
metrics = {}
# 计算成交量均值
volume_window = SINGLE_STOCK_CONFIG["volume"]["volume_window"]
if len(kline_data) >= volume_window:
avg_volume = kline_data["volume"].rolling(
window=volume_window).mean()
current_volume = kline_data["volume"].iloc[-1]
metrics["current_volume"] = current_volume
metrics["avg_volume"] = avg_volume.iloc[-1]
volume_increase_ratio = SINGLE_STOCK_CONFIG[
"volume"]["volume_increase_ratio"]
volume_decrease_ratio = SINGLE_STOCK_CONFIG[
"volume"]["volume_decrease_ratio"]
if current_volume > avg_volume.iloc[-1] * volume_increase_ratio:
score += 10
reasons.append("成交量放大，市场关注增加")
elif current_volume < avg_volume.iloc[-1] * volume_decrease_ratio:
score += 2
reasons.append("成交量萎缩，市场关注减少")
else:
score += 5
reasons.append("成交量正常")
# 应用权重
score *= SINGLE_STOCK_CONFIG["volume"]["volume_score_weight"]
return score, reasons, metrics
def _analyze_risk(self,
kline_data: pd.DataFrame) -> (float,
List[str],
Dict[str, float]):
"""
分析风险
Args:
kline_data: K 线数据
Returns:
(风险评估得分, 风险评估理由, 风险指标数值)
"""
score = 0.0
reasons = []
metrics = {}
# 计算波动率
if len(kline_data) >= 10:
returns = kline_data["close"].pct_change()
volatility = returns.rolling(
window=10).std() * (252 ** 0.5) # 年化波动率
metrics["volatility"] = volatility.iloc[-1]
if volatility.iloc[-1] < 0.3:
score += 10
reasons.append("波动率较低，风险可控")
elif volatility.iloc[-1] < 0.5:
score += 5
reasons.append("波动率适中，风险中等")
else:
score += 2
reasons.append("波动率较高，风险较大")
# 计算最大回撤
if len(kline_data) >= 20:
roll_max = kline_data["close"].rolling(window=20).max()
daily_drawdown = kline_data["close"] / roll_max - 1.0
max_drawdown = daily_drawdown.rolling(window=20).min()
metrics["max_drawdown"] = max_drawdown.iloc[-1]
if max_drawdown.iloc[-1] > -0.1:
score += 10
reasons.append("最大回撤较小，风险可控")
elif max_drawdown.iloc[-1] > -0.2:
score += 5
reasons.append("最大回撤适中，风险中等")
else:
score += 2
reasons.append("最大回撤较大，风险较大")
return score, reasons, metrics
def _calculate_rating(self, score: float) -> str:
"""
根据评分计算标的等级
Args:
score: 综合评分
Returns:
标的等级（S/A/B/C）
"""
if score >= 85:
return "S"
elif score >= 70:
return "A"
elif score >= 50:
return "B"
else:
return "C"
def _determine_trading_signal(self,
score: float,
kline_data: pd.DataFrame) -> str:
"""
确定交易信号
Args:
score: 综合评分
kline_data: K 线数据
Returns:
交易信号（买入/持有/卖出）
"""
if score >= 50:
return "买入"
elif score >= 40:
return "持有"
else:
return "卖出"