|
|
# 压力支撑模块
|
|
|
import pandas as pd
|
|
|
import numpy as np
|
|
|
from typing import Dict, List, Tuple, Optional
|
|
|
from qihuo_analyzer.utils.technical_analysis import calculate_bollinger_bands
|
|
|
|
|
|
|
|
|
class SupportResistance:
|
|
|
"""压力支撑分析器"""
|
|
|
|
|
|
def __init__(self):
|
|
|
pass
|
|
|
|
|
|
def analyze_support_resistance(self, data: pd.DataFrame) -> Dict:
|
|
|
"""分析压力支撑位"""
|
|
|
result = {}
|
|
|
|
|
|
# 识别关键价位
|
|
|
key_levels = self._identify_key_levels(data)
|
|
|
result.update(key_levels)
|
|
|
|
|
|
# 计算枢轴点
|
|
|
pivot_points = self._calculate_pivot_points(data)
|
|
|
result.update(pivot_points)
|
|
|
|
|
|
# 基于布林带的支撑阻力
|
|
|
bollinger_levels = self._calculate_bollinger_levels(data)
|
|
|
result.update(bollinger_levels)
|
|
|
|
|
|
# 最近高低点分析
|
|
|
recent_high_low = self._analyze_recent_high_low(data)
|
|
|
result.update(recent_high_low)
|
|
|
|
|
|
# 斐波那契回调线
|
|
|
fibonacci_levels = self._calculate_fibonacci_levels(data)
|
|
|
result['fibonacci_levels'] = fibonacci_levels
|
|
|
|
|
|
# 综合支撑阻力位
|
|
|
support_resistance_levels = self._generate_support_resistance_levels(result)
|
|
|
result['support_resistance_levels'] = support_resistance_levels
|
|
|
|
|
|
return result
|
|
|
|
|
|
def _identify_key_levels(self, data: pd.DataFrame) -> Dict:
|
|
|
"""识别关键价位"""
|
|
|
# 计算最近N天的高低点
|
|
|
recent_high = data['high'].tail(50).max()
|
|
|
recent_low = data['low'].tail(50).min()
|
|
|
|
|
|
# 计算最近N天的平均波幅
|
|
|
avg_range = (data['high'] - data['low']).tail(50).mean()
|
|
|
|
|
|
# 识别成交量密集区
|
|
|
volume_profile = self._calculate_volume_profile(data)
|
|
|
|
|
|
# 识别价格密集区
|
|
|
price_density = self._calculate_price_density(data)
|
|
|
|
|
|
return {
|
|
|
'recent_high': recent_high,
|
|
|
'recent_low': recent_low,
|
|
|
'avg_range': avg_range,
|
|
|
'volume_profile': volume_profile,
|
|
|
'price_density': price_density
|
|
|
}
|
|
|
|
|
|
def _calculate_pivot_points(self, data: pd.DataFrame) -> Dict:
|
|
|
"""计算枢轴点"""
|
|
|
# 使用最近的高点、低点和收盘价计算枢轴点
|
|
|
if len(data) < 2:
|
|
|
return {
|
|
|
'pivot_point': None,
|
|
|
'resistance_1': None,
|
|
|
'resistance_2': None,
|
|
|
'support_1': None,
|
|
|
'support_2': None
|
|
|
}
|
|
|
|
|
|
high = data['high'].iloc[-1]
|
|
|
low = data['low'].iloc[-1]
|
|
|
close = data['close'].iloc[-1]
|
|
|
|
|
|
# 计算枢轴点
|
|
|
pivot_point = (high + low + close) / 3
|
|
|
|
|
|
# 计算阻力位和支撑位
|
|
|
resistance_1 = 2 * pivot_point - low
|
|
|
resistance_2 = pivot_point + (high - low)
|
|
|
support_1 = 2 * pivot_point - high
|
|
|
support_2 = pivot_point - (high - low)
|
|
|
|
|
|
return {
|
|
|
'pivot_point': pivot_point,
|
|
|
'resistance_1': resistance_1,
|
|
|
'resistance_2': resistance_2,
|
|
|
'support_1': support_1,
|
|
|
'support_2': support_2
|
|
|
}
|
|
|
|
|
|
def _calculate_bollinger_levels(self, data: pd.DataFrame) -> Dict:
|
|
|
"""基于布林带的支撑阻力"""
|
|
|
# 计算布林带
|
|
|
bollinger_data = calculate_bollinger_bands(data)
|
|
|
|
|
|
# 获取最新的布林带值
|
|
|
upper_band = bollinger_data['upper_band'].iloc[-1]
|
|
|
middle_band = bollinger_data['sma'].iloc[-1]
|
|
|
lower_band = bollinger_data['lower_band'].iloc[-1]
|
|
|
|
|
|
return {
|
|
|
'bollinger_upper': upper_band,
|
|
|
'bollinger_middle': middle_band,
|
|
|
'bollinger_lower': lower_band
|
|
|
}
|
|
|
|
|
|
def _analyze_recent_high_low(self, data: pd.DataFrame) -> Dict:
|
|
|
"""最近高低点分析"""
|
|
|
# 计算不同周期的高低点
|
|
|
periods = [10, 20, 50]
|
|
|
high_low_levels = {}
|
|
|
|
|
|
for period in periods:
|
|
|
if len(data) >= period:
|
|
|
high_low_levels[f'{period}d_high'] = data['high'].tail(period).max()
|
|
|
high_low_levels[f'{period}d_low'] = data['low'].tail(period).min()
|
|
|
else:
|
|
|
high_low_levels[f'{period}d_high'] = None
|
|
|
high_low_levels[f'{period}d_low'] = None
|
|
|
|
|
|
return high_low_levels
|
|
|
|
|
|
def _calculate_volume_profile(self, data: pd.DataFrame) -> Dict:
|
|
|
"""计算成交量分布"""
|
|
|
# 简化的成交量分布分析
|
|
|
# 将价格区间分成10个区间，计算每个区间的成交量
|
|
|
if len(data) < 10:
|
|
|
return {}
|
|
|
|
|
|
price_min = data['low'].tail(50).min()
|
|
|
price_max = data['high'].tail(50).max()
|
|
|
price_range = price_max - price_min
|
|
|
bin_size = price_range / 10
|
|
|
|
|
|
volume_profile = {}
|
|
|
for i in range(10):
|
|
|
bin_low = price_min + i * bin_size
|
|
|
bin_high = price_min + (i + 1) * bin_size
|
|
|
|
|
|
# 计算该价格区间的成交量
|
|
|
bin_volume = data[
|
|
|
(data['low'] <= bin_high) &
|
|
|
(data['high'] >= bin_low)
|
|
|
]['volume'].sum()
|
|
|
|
|
|
volume_profile[f'bin_{i+1}'] = {
|
|
|
'price_range': (bin_low, bin_high),
|
|
|
'volume': bin_volume
|
|
|
}
|
|
|
|
|
|
# 找出成交量最大的区间
|
|
|
max_volume_bin = max(volume_profile.items(), key=lambda x: x[1]['volume'])
|
|
|
|
|
|
return {
|
|
|
'volume_profile': volume_profile,
|
|
|
'max_volume_bin': max_volume_bin
|
|
|
}
|
|
|
|
|
|
def _calculate_price_density(self, data: pd.DataFrame) -> Dict:
|
|
|
"""计算价格密度"""
|
|
|
# 简化的价格密度分析
|
|
|
if len(data) < 10:
|
|
|
return {}
|
|
|
|
|
|
# 计算收盘价的分布
|
|
|
prices = data['close'].tail(100)
|
|
|
price_std = prices.std()
|
|
|
price_mean = prices.mean()
|
|
|
|
|
|
# 计算价格分位数
|
|
|
price_percentiles = {
|
|
|
'p10': np.percentile(prices, 10),
|
|
|
'p25': np.percentile(prices, 25),
|
|
|
'p50': np.percentile(prices, 50),
|
|
|
'p75': np.percentile(prices, 75),
|
|
|
'p90': np.percentile(prices, 90)
|
|
|
}
|
|
|
|
|
|
return {
|
|
|
'price_mean': price_mean,
|
|
|
'price_std': price_std,
|
|
|
'price_percentiles': price_percentiles
|
|
|
}
|
|
|
|
|
|
def _calculate_fibonacci_levels(self, data: pd.DataFrame) -> Dict:
|
|
|
"""计算斐波那契回调线"""
|
|
|
if len(data) < 20:
|
|
|
return {}
|
|
|
|
|
|
# 找出最近的显著高低点
|
|
|
swing_high = data['high'].tail(50).max()
|
|
|
swing_low = data['low'].tail(50).min()
|
|
|
|
|
|
# 计算斐波那契回调位
|
|
|
range_high_low = swing_high - swing_low
|
|
|
|
|
|
fib_levels = {
|
|
|
'0': swing_low,
|
|
|
'0.236': swing_low + range_high_low * 0.236,
|
|
|
'0.382': swing_low + range_high_low * 0.382,
|
|
|
'0.5': swing_low + range_high_low * 0.5,
|
|
|
'0.618': swing_low + range_high_low * 0.618,
|
|
|
'0.786': swing_low + range_high_low * 0.786,
|
|
|
'1': swing_high
|
|
|
}
|
|
|
|
|
|
return fib_levels
|
|
|
|
|
|
def _generate_support_resistance_levels(self, analysis: Dict) -> Dict:
|
|
|
"""生成综合支撑阻力位"""
|
|
|
# 收集所有可能的支撑阻力位
|
|
|
all_levels = []
|
|
|
|
|
|
# 添加最近高低点
|
|
|
all_levels.append(analysis.get('recent_high', 0))
|
|
|
all_levels.append(analysis.get('recent_low', 0))
|
|
|
|
|
|
# 添加枢轴点相关价位
|
|
|
all_levels.extend([
|
|
|
analysis.get('pivot_point', 0),
|
|
|
analysis.get('resistance_1', 0),
|
|
|
analysis.get('resistance_2', 0),
|
|
|
analysis.get('support_1', 0),
|
|
|
analysis.get('support_2', 0)
|
|
|
])
|
|
|
|
|
|
# 添加布林带相关价位
|
|
|
all_levels.extend([
|
|
|
analysis.get('bollinger_upper', 0),
|
|
|
analysis.get('bollinger_middle', 0),
|
|
|
analysis.get('bollinger_lower', 0)
|
|
|
])
|
|
|
|
|
|
# 添加不同周期的高低点
|
|
|
periods = [10, 20, 50]
|
|
|
for period in periods:
|
|
|
all_levels.append(analysis.get(f'{period}d_high', 0))
|
|
|
all_levels.append(analysis.get(f'{period}d_low', 0))
|
|
|
|
|
|
# 添加斐波那契回调位
|
|
|
fib_levels = analysis.get('fibonacci_levels', {})
|
|
|
all_levels.extend(fib_levels.values())
|
|
|
|
|
|
# 过滤无效值并排序
|
|
|
all_levels = [level for level in all_levels if level and level > 0]
|
|
|
all_levels.sort()
|
|
|
|
|
|
# 去重（相近的价位视为同一价位）
|
|
|
if not all_levels:
|
|
|
return {'support_levels': [], 'resistance_levels': []}
|
|
|
|
|
|
unique_levels = []
|
|
|
threshold = analysis.get('avg_range', 10) * 0.3 # 阈值为平均波幅的30%
|
|
|
|
|
|
for level in all_levels:
|
|
|
if not unique_levels or abs(level - unique_levels[-1]) > threshold:
|
|
|
unique_levels.append(level)
|
|
|
|
|
|
# 确定当前价格
|
|
|
current_price = analysis.get('recent_high', 3500) * 0.95 # 使用最近高点的95%作为当前价格
|
|
|
|
|
|
# 分离支撑位和阻力位
|
|
|
support_levels = [level for level in unique_levels if level < current_price]
|
|
|
resistance_levels = [level for level in unique_levels if level > current_price]
|
|
|
|
|
|
# 按距离当前价格排序
|
|
|
support_levels.sort(reverse=True) # 最近的支撑位在前
|
|
|
resistance_levels.sort() # 最近的阻力位在前
|
|
|
|
|
|
# 取最近的几个支撑阻力位
|
|
|
support_levels = support_levels[:3] # 最近的3个支撑位
|
|
|
resistance_levels = resistance_levels[:3] # 最近的3个阻力位
|
|
|
|
|
|
return {
|
|
|
'support_levels': support_levels,
|
|
|
'resistance_levels': resistance_levels,
|
|
|
'current_price': current_price
|
|
|
}
|
|
|
|
|
|
def calculate_stop_loss_level(self, data: pd.DataFrame, direction: str, atr: float) -> float:
|
|
|
"""计算智能止损位"""
|
|
|
# 分析支撑阻力位
|
|
|
sr_analysis = self.analyze_support_resistance(data)
|
|
|
support_levels = sr_analysis.get('support_resistance_levels', {}).get('support_levels', [])
|
|
|
resistance_levels = sr_analysis.get('support_resistance_levels', {}).get('resistance_levels', [])
|
|
|
current_price = data['close'].iloc[-1]
|
|
|
|
|
|
if direction == 'long':
|
|
|
# 做多时，止损位应在最近的支撑位下方
|
|
|
if support_levels:
|
|
|
# 最近的支撑位下方ATR的0.5倍
|
|
|
stop_loss = support_levels[0] - atr * 0.5
|
|
|
else:
|
|
|
# 没有支撑位时，使用ATR的2倍
|
|
|
stop_loss = current_price - atr * 2
|
|
|
elif direction == 'short':
|
|
|
# 做空时，止损位应在最近的阻力位上方
|
|
|
if resistance_levels:
|
|
|
# 最近的阻力位上方ATR的0.5倍
|
|
|
stop_loss = resistance_levels[0] + atr * 0.5
|
|
|
else:
|
|
|
# 没有阻力位时，使用ATR的2倍
|
|
|
stop_loss = current_price + atr * 2
|
|
|
else:
|
|
|
raise ValueError("Direction must be 'long' or 'short'")
|
|
|
|
|
|
return stop_loss
|
|
|
|
|
|
def calculate_target_price(self, data: pd.DataFrame, direction: str, entry_price: float) -> float:
|
|
|
"""计算目标价"""
|
|
|
# 分析支撑阻力位
|
|
|
sr_analysis = self.analyze_support_resistance(data)
|
|
|
support_levels = sr_analysis.get('support_resistance_levels', {}).get('support_levels', [])
|
|
|
resistance_levels = sr_analysis.get('support_resistance_levels', {}).get('resistance_levels', [])
|
|
|
|
|
|
if direction == 'long':
|
|
|
# 做多时，目标价应在最近的阻力位
|
|
|
if resistance_levels:
|
|
|
target_price = resistance_levels[0]
|
|
|
else:
|
|
|
# 没有阻力位时，使用近期高点
|
|
|
target_price = sr_analysis.get('recent_high', entry_price * 1.05)
|
|
|
elif direction == 'short':
|
|
|
# 做空时，目标价应在最近的支撑位
|
|
|
if support_levels:
|
|
|
target_price = support_levels[0]
|
|
|
else:
|
|
|
# 没有支撑位时，使用近期低点
|
|
|
target_price = sr_analysis.get('recent_low', entry_price * 0.95)
|
|
|
else:
|
|
|
raise ValueError("Direction must be 'long' or 'short'")
|
|
|
|
|
|
return target_price
|
|
|
|
|
|
def analyze_price_position(self, data: pd.DataFrame) -> Dict:
|
|
|
"""分析价格位置"""
|
|
|
current_price = data['close'].iloc[-1]
|
|
|
|
|
|
# 分析支撑阻力位
|
|
|
sr_analysis = self.analyze_support_resistance(data)
|
|
|
support_levels = sr_analysis.get('support_resistance_levels', {}).get('support_levels', [])
|
|
|
resistance_levels = sr_analysis.get('support_resistance_levels', {}).get('resistance_levels', [])
|
|
|
|
|
|
# 计算价格与支撑阻力位的距离
|
|
|
distance_to_support = float('inf')
|
|
|
distance_to_resistance = float('inf')
|
|
|
|
|
|
if support_levels:
|
|
|
distance_to_support = current_price - support_levels[0]
|
|
|
|
|
|
if resistance_levels:
|
|
|
distance_to_resistance = resistance_levels[0] - current_price
|
|
|
|
|
|
# 分析价格位置
|
|
|
position = 'neutral'
|
|
|
if distance_to_resistance < sr_analysis.get('avg_range', 10) * 0.2:
|
|
|
position = 'near_resistance'
|
|
|
elif distance_to_support < sr_analysis.get('avg_range', 10) * 0.2:
|
|
|
position = 'near_support'
|
|
|
|
|
|
# 分析价格在布林带中的位置
|
|
|
bollinger_upper = sr_analysis.get('bollinger_upper', 0)
|
|
|
bollinger_middle = sr_analysis.get('bollinger_middle', 0)
|
|
|
bollinger_lower = sr_analysis.get('bollinger_lower', 0)
|
|
|
|
|
|
bollinger_position = 'middle'
|
|
|
if current_price > bollinger_upper:
|
|
|
bollinger_position = 'upper'
|
|
|
elif current_price < bollinger_lower:
|
|
|
bollinger_position = 'lower'
|
|
|
|
|
|
return {
|
|
|
'current_price': current_price,
|
|
|
'distance_to_support': distance_to_support,
|
|
|
'distance_to_resistance': distance_to_resistance,
|
|
|
'position': position,
|
|
|
'bollinger_position': bollinger_position,
|
|
|
'support_levels': support_levels,
|
|
|
'resistance_levels': resistance_levels
|
|
|
}
|
