strategy_analyzer.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
策略分析模块
包含独立的单只股票分析函数，与外层遍历主程序完全解耦
策略逻辑仅负责单票的分析计算，不涉及股票池遍历和批量数据获取
"""
import os
import logging
import pandas as pd
import numpy as np
import json
import time
from pathlib import Path
from datetime import datetime, timedelta
from dotenv import load_dotenv
from src.config_manager import config_manager
from src.utils import ensure_dir_exists, load_json_file, save_json_file, get_trade_date
# 加载环境变量
load_dotenv()
# 配置日志
logger = logging.getLogger(__name__)
class StrategyAnalyzer:
"""策略分析器类"""
def __init__(self, strategy_name='model'):
"""初始化策略分析器"""
self.strategy_name = strategy_name
self.params = self._load_strategy_parameters()
logger.info(f"策略分析器初始化成功，策略：{strategy_name}")
def _load_strategy_parameters(self):
"""加载策略参数配置（可自定义）"""
params_file = f"config/{self.strategy_name}_strategy.json"
if Path(params_file).exists():
try:
with open(params_file, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
logger.error(f"加载策略参数失败：{e}")
# 默认参数（可根据需要修改）
return {
"learning_window": "90日",
"profit_threshold": "3%",
"stock_count": "10只",
"volume_ratio_threshold": "1.5倍"
}
def analyze_single_stock(self, ts_code, df_stock):
"""
单只股票分析函数（策略核心逻辑，与主程序解耦）
参数：
ts_code: str - 股票代码
df_stock: pd.DataFrame - 单只股票的完整K线/指标数据
返回：
dict - 结构化分析结果，包含：
ts_code: 股票代码
score: 综合评分（0-100分）
reason: 评分理由（分点明确关键指标/形态的具体数值和判断依据）
level: 标的等级（S/A/B/C）
signals: 核心信号关键点
"""
try:
logger.info(f"开始分析股票：{ts_code}")
# 初始化分析结果
analysis_result = {
"ts_code": ts_code,
"score": 0,
"reason": "",
"level": "C",
"signals": []
}
# 检查数据是否有效
if df_stock.empty:
analysis_result["reason"] = "无有效数据"
return analysis_result
# 根据策略名称选择分析方法
if self.strategy_name == 'model':
analysis_result = self._analyze_model_strategy(analysis_result, df_stock)
elif self.strategy_name == 'vcp':
analysis_result = self._analyze_vcp_strategy(analysis_result, df_stock)
elif self.strategy_name == 'test':
analysis_result = self._analyze_test_strategy(analysis_result, df_stock)
else:
analysis_result["reason"] = "不支持的策略"
return analysis_result
# 根据评分确定标的等级
analysis_result["level"] = self._determine_level(analysis_result["score"])
logger.info(f"股票分析完成：{ts_code}，评分：{analysis_result['score']:.1f}，等级：{analysis_result['level']}")
return analysis_result
except Exception as e:
logger.error(f"分析股票 {ts_code} 失败：{e}")
return {
"ts_code": ts_code,
"score": 0,
"reason": f"分析失败：{str(e)}",
"level": "C",
"signals": []
}
def _analyze_model_strategy(self, analysis_result, df_stock):
"""大模型选股策略分析（可自定义）"""
try:
# 计算基本指标
latest_close = df_stock['close'].iloc[-1]
latest_volume = df_stock['vol'].iloc[-1]
latest_turnover = df_stock['amount_x'].iloc[-1]
latest_pct_chg = df_stock['pct_chg'].iloc[-1]
# 计算移动平均线
ma5 = df_stock['close'].rolling(window=5).mean().iloc[-1]
ma10 = df_stock['close'].rolling(window=10).mean().iloc[-1]
ma20 = df_stock['close'].rolling(window=20).mean().iloc[-1]
# 计算成交量相关指标
avg_volume = df_stock['vol'].rolling(window=20).mean().iloc[-1]
volume_ratio = latest_volume / avg_volume if avg_volume != 0 and not np.isnan(avg_volume) else 0
# 计算价格波动指标
volatility = df_stock['close'].pct_change().rolling(window=20).std().iloc[-1]
# 评分逻辑（可自定义）
score = 0
reasons = []
# 价格趋势评分（20分）
if latest_close > ma5 and ma5 > ma10 and ma10 > ma20:
score += 20
reasons.append("价格呈上升趋势，MA5>MA10>MA20")
# 成交量评分（20分）
if volume_ratio > 1.5 and not np.isnan(volume_ratio):
score += 20
reasons.append(f"成交量放大，量比 {volume_ratio:.1f} 倍")
elif volume_ratio > 1.0 and not np.isnan(volume_ratio):
score += 10
reasons.append(f"成交量正常，量比 {volume_ratio:.1f} 倍")
elif not np.isnan(volume_ratio):
reasons.append(f"成交量不足，量比 {volume_ratio:.1f} 倍")
# 涨幅评分（20分）
if latest_pct_chg > 3:
score += 20
reasons.append(f"当日涨幅显著，{latest_pct_chg:.2f}%")
elif latest_pct_chg > 1:
score += 10
reasons.append(f"当日小幅上涨，{latest_pct_chg:.2f}%")
# 价格波动评分（20分）
if volatility < 0.05 and not np.isnan(volatility):
score += 20
reasons.append(f"价格波动小，波动率 {volatility:.1%}")
elif volatility < 0.10 and not np.isnan(volatility):
score += 10
reasons.append(f"价格波动适中，波动率 {volatility:.1%}")
elif not np.isnan(volatility):
reasons.append(f"价格波动大，波动率 {volatility:.1%}")
# 换手率评分（20分）
# 使用成交量计算换手率（换手率=成交量/流通股本）
# 简单估算流通股本为成交量的 1000000 倍（实际应该从数据接口获取）
if latest_volume > 0 and not np.isnan(latest_volume):
estimated_circulation = 100000000 # 估算流通股本为 1 亿股
turnover_rate = latest_volume / estimated_circulation
if turnover_rate > 0.02:
score += 20
reasons.append(f"换手率高，{turnover_rate:.2%}")
elif turnover_rate > 0.01:
score += 10
reasons.append(f"换手率适中，{turnover_rate:.2%}")
else:
reasons.append(f"换手率低，{turnover_rate:.2%}")
# 信号关键点
signals = []
if volume_ratio > 1.5 and latest_pct_chg > 3:
signals.append("放量上涨")
if volatility < 0.05:
signals.append("价格稳定")
# 构建分析结果
analysis_result["score"] = min(score, 100)
analysis_result["reason"] = "; ".join(reasons)
analysis_result["signals"] = signals
return analysis_result
except Exception as e:
logger.error(f"大模型策略分析失败：{e}")
analysis_result["reason"] = f"策略分析失败：{str(e)}"
return analysis_result
def _analyze_vcp_strategy(self, analysis_result, df_stock):
"""VCP极致坍塌模型策略分析（可自定义）"""
try:
# 计算价格波动率
df_stock['volatility'] = df_stock['close'].pct_change().rolling(20).std()
# 寻找波动率收缩的时期
contraction_periods = self._find_volatility_contraction(df_stock['volatility'])
# 评分逻辑（可自定义）
score = 0
reasons = []
if contraction_periods:
latest_contraction = contraction_periods[-1]
# 收缩期长度评分（30分）
if latest_contraction['length'] > 20 and latest_contraction['length'] < 60:
score += 30
reasons.append(f"收缩期长度适中，{latest_contraction['length']}个交易日")
# 波动率收缩幅度评分（30分）
if latest_contraction['contraction_ratio'] > 0.15:
score += 30
reasons.append(f"波动率收缩显著，{latest_contraction['contraction_ratio']:.1%}")
# 突破强度评分（40分）
breakout_range = self._calculate_breakout_range(df_stock, latest_contraction['end_index'])
if breakout_range > 0.05:
score += 40
reasons.append(f"价格突破强度高，{breakout_range:.1%}")
# 信号关键点
signals = []
if contraction_periods:
signals.append("VCP形态识别成功")
if latest_contraction['contraction_ratio'] > 0.15:
signals.append("波动率显著收缩")
# 构建分析结果
analysis_result["score"] = min(score, 100)
analysis_result["reason"] = "; ".join(reasons) if reasons else "未符合VCP形态特征"
analysis_result["signals"] = signals
return analysis_result
except Exception as e:
logger.error(f"VCP策略分析失败：{e}")
analysis_result["reason"] = f"策略分析失败：{str(e)}"
return analysis_result
def _analyze_test_strategy(self, analysis_result, df_stock):
"""测试选股策略分析（可自定义）"""
try:
# 计算基本指标
latest_close = df_stock['close'].iloc[-1]
latest_volume = df_stock['vol'].iloc[-1]
# 计算价格波动指标
volatility = df_stock['close'].pct_change().rolling(20).std().iloc[-1]
# 计算RSI指标
rsi = self._calculate_rsi(df_stock['close'], 14).iloc[-1]
# 计算MACD指标
macd, _, _ = self._calculate_macd(df_stock['close'])
# 评分逻辑（可自定义）
score = 0
reasons = []
# 价格范围评分（25分）
if latest_close > 10 and latest_close < 100:
score += 25
reasons.append(f"价格在合理范围，{latest_close:.2f}")
# 成交量评分（25分）
avg_volume = df_stock['vol'].rolling(window=20).mean().iloc[-1]
volume_ratio = latest_volume / avg_volume if avg_volume != 0 else 0
if volume_ratio > 1.5:
score += 25
reasons.append(f"成交量放大，量比 {volume_ratio:.1f} 倍")
# 波动率评分（25分）
if volatility < 0.05:
score += 25
reasons.append(f"价格波动小，波动率 {volatility:.1%}")
# RSI评分（25分）
if rsi > 30 and rsi < 70:
score += 25
reasons.append(f"RSI指标正常，{rsi:.1f}")
# 信号关键点
signals = []
if volume_ratio > 1.5:
signals.append("放量")
if volatility < 0.05:
signals.append("价格稳定")
# 构建分析结果
analysis_result["score"] = min(score, 100)
analysis_result["reason"] = "; ".join(reasons) if reasons else "未符合测试策略条件"
analysis_result["signals"] = signals
return analysis_result
except Exception as e:
logger.error(f"测试策略分析失败：{e}")
analysis_result["reason"] = f"策略分析失败：{str(e)}"
return analysis_result
def _determine_level(self, score):
"""根据评分确定标的等级（可自定义）"""
if score >= 80:
return "S"
elif score >= 60:
return "A"
elif score >= 40:
return "B"
else:
return "C"
def _find_volatility_contraction(self, volatility_series):
"""寻找波动率收缩的时期"""
contraction_periods = []
start_idx = None
for i in range(1, len(volatility_series)):
if start_idx is None and volatility_series[i] < volatility_series[i-1]:
start_idx = i-1
if start_idx is not None and volatility_series[i] > volatility_series[i-1]:
contraction_length = i - start_idx
contraction_ratio = (volatility_series[start_idx] - volatility_series[i-1]) / volatility_series[start_idx]
contraction_periods.append({
"start_index": start_idx,
"end_index": i-1,
"length": contraction_length,
"contraction_ratio": contraction_ratio,
"start_vol": volatility_series[start_idx],
"end_vol": volatility_series[i-1]
})
start_idx = None
return contraction_periods
def _calculate_breakout_range(self, df_stock, end_index):
"""计算突破幅度"""
try:
breakout_data = df_stock.iloc[end_index:end_index+5]
max_price = breakout_data['high'].max()
min_price = breakout_data['low'].min()
breakout_range = (max_price - min_price) / min_price
return breakout_range
except Exception as e:
logger.error(f"计算突破幅度失败：{e}")
return 0
def _calculate_rsi(self, prices, period=14):
"""计算RSI指标"""
delta = prices.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_macd(self, prices, fast=12, slow=26, signal_period=9):
"""计算MACD指标"""
ema_fast = prices.ewm(span=fast, adjust=False).mean()
ema_slow = prices.ewm(span=slow, adjust=False).mean()
macd = ema_fast - ema_slow
signal = macd.ewm(span=signal_period, adjust=False).mean()
hist = macd - signal
return macd.values, signal.values, hist.values
# 策略分析器实例（可根据需要创建不同策略的分析器）
model_analyzer = StrategyAnalyzer(strategy_name='model')
vcp_analyzer = StrategyAnalyzer(strategy_name='vcp')
test_analyzer = StrategyAnalyzer(strategy_name='test')