683 lines
26 KiB
Python
683 lines
26 KiB
Python
"""
|
||
分型识别模块：识别顶分型和底分型
|
||
分型定义：至少需要3根K线，中间K线的高点（或低点）比两侧都高（或低）
|
||
增强版：包含多维度强弱程度评估
|
||
"""
|
||
|
||
import pandas as pd
|
||
import numpy as np
|
||
from typing import List, Tuple, Optional, Dict
|
||
from dataclasses import dataclass
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class FractalPoint:
|
||
"""分型点数据类"""
|
||
index: int # 在原始数据中的位置
|
||
timestamp: pd.Timestamp # 时间戳
|
||
price: float # 分型价格（高点或低点）
|
||
fractal_type: str # 'top' 或 'bottom'
|
||
strength: int # 基础分型强度（左右确认的K线数量）
|
||
enhanced_strength: float # 增强强度评分（0-100）
|
||
price_dominance: float # 价格优势度（0-100）
|
||
volume_strength: float # 成交量强度（0-100）
|
||
trend_position: float # 趋势位置强度（0-100）
|
||
confirmed: bool = False # 是否已确认
|
||
|
||
|
||
class Fractal:
|
||
"""分型识别类（增强版）"""
|
||
|
||
def __init__(self, kline_data: pd.DataFrame, min_strength: int = 1):
|
||
"""
|
||
初始化分型识别器
|
||
|
||
Args:
|
||
kline_data: 处理包含关系后的K线数据
|
||
min_strength: 最小分型强度（左右各需多少根K线确认）
|
||
"""
|
||
self.data = kline_data.copy()
|
||
self.min_strength = max(1, min_strength) # 至少为1
|
||
self.top_fractals = []
|
||
self.bottom_fractals = []
|
||
self.all_fractals = []
|
||
|
||
# 计算一些辅助指标
|
||
self._calculate_auxiliary_indicators()
|
||
|
||
def _calculate_auxiliary_indicators(self):
|
||
"""计算辅助技术指标"""
|
||
# 成交量移动平均
|
||
self.data['volume_ma'] = self.data['volume'].rolling(window=20, min_periods=1).mean()
|
||
|
||
# 价格振幅
|
||
self.data['range'] = self.data['high'] - self.data['low']
|
||
self.data['range_ma'] = self.data['range'].rolling(window=20, min_periods=1).mean()
|
||
|
||
# 相对位置（高点在整个区间的位置）
|
||
window = 20
|
||
self.data['highest'] = self.data['high'].rolling(window=window, min_periods=1).max()
|
||
self.data['lowest'] = self.data['low'].rolling(window=window, min_periods=1).min()
|
||
|
||
# 趋势强度（简单移动平均斜率）
|
||
self.data['close_ma'] = self.data['close'].rolling(window=10, min_periods=1).mean()
|
||
self.data['trend_slope'] = self.data['close_ma'].diff(5) # 5期斜率
|
||
|
||
def _calculate_price_dominance(self, idx: int, fractal_type: str, strength: int) -> float:
|
||
"""
|
||
计算价格优势度：分型价格相对于周围价格的优势程度
|
||
|
||
Args:
|
||
idx: 分型位置
|
||
fractal_type: 分型类型
|
||
strength: 基础强度
|
||
|
||
Returns:
|
||
价格优势度（0-100）
|
||
"""
|
||
try:
|
||
# 扩展检查范围
|
||
check_range = max(strength * 2, 10)
|
||
start_idx = max(0, idx - check_range)
|
||
end_idx = min(len(self.data), idx + check_range + 1)
|
||
|
||
if fractal_type == 'top':
|
||
center_price = self.data.iloc[idx]['high']
|
||
range_data = self.data.iloc[start_idx:end_idx]
|
||
max_around = range_data['high'].max()
|
||
second_max = range_data['high'].nlargest(2).iloc[1] if len(range_data) > 1 else center_price
|
||
|
||
# 计算相对优势
|
||
if max_around == center_price:
|
||
price_gap = center_price - second_max
|
||
avg_range = self.data.iloc[idx]['range_ma']
|
||
dominance = min((price_gap / avg_range) * 50, 100) if avg_range > 0 else 50
|
||
else:
|
||
dominance = 0
|
||
else: # bottom
|
||
center_price = self.data.iloc[idx]['low']
|
||
range_data = self.data.iloc[start_idx:end_idx]
|
||
min_around = range_data['low'].min()
|
||
second_min = range_data['low'].nsmallest(2).iloc[1] if len(range_data) > 1 else center_price
|
||
|
||
# 计算相对优势
|
||
if min_around == center_price:
|
||
price_gap = second_min - center_price
|
||
avg_range = self.data.iloc[idx]['range_ma']
|
||
dominance = min((price_gap / avg_range) * 50, 100) if avg_range > 0 else 50
|
||
else:
|
||
dominance = 0
|
||
|
||
return max(0, dominance)
|
||
|
||
except Exception as e:
|
||
logger.warning(f"计算价格优势度失败: {e}")
|
||
return 30 # 默认值
|
||
|
||
def _calculate_volume_strength(self, idx: int, fractal_type: str, strength: int) -> float:
|
||
"""
|
||
计算成交量强度：分型形成时的成交量特征
|
||
|
||
Args:
|
||
idx: 分型位置
|
||
fractal_type: 分型类型
|
||
strength: 基础强度
|
||
|
||
Returns:
|
||
成交量强度（0-100）
|
||
"""
|
||
try:
|
||
center_volume = self.data.iloc[idx]['volume']
|
||
volume_ma = self.data.iloc[idx]['volume_ma']
|
||
|
||
# 基础成交量比率
|
||
volume_ratio = center_volume / volume_ma if volume_ma > 0 else 1
|
||
base_score = min(volume_ratio * 30, 60) # 最高60分
|
||
|
||
# 检查分型形成过程中的成交量模式
|
||
pattern_score = 0
|
||
if strength >= 2:
|
||
# 检查左右成交量是否递减（表示力度衰竭）
|
||
left_volumes = [self.data.iloc[idx - i]['volume'] for i in range(1, strength + 1)]
|
||
right_volumes = [self.data.iloc[idx + i]['volume'] for i in range(1, strength + 1)]
|
||
|
||
# 中心成交量应该相对突出
|
||
center_prominence = sum([1 for v in left_volumes + right_volumes if center_volume > v])
|
||
total_compared = len(left_volumes + right_volumes)
|
||
pattern_score = (center_prominence / total_compared) * 40 if total_compared > 0 else 20
|
||
|
||
total_score = base_score + pattern_score
|
||
return max(0, min(total_score, 100))
|
||
|
||
except Exception as e:
|
||
logger.warning(f"计算成交量强度失败: {e}")
|
||
return 40 # 默认值
|
||
|
||
def _calculate_trend_position_strength(self, idx: int, fractal_type: str) -> float:
|
||
"""
|
||
计算趋势位置强度：分型在整体趋势中的位置优势
|
||
|
||
Args:
|
||
idx: 分型位置
|
||
fractal_type: 分型类型
|
||
|
||
Returns:
|
||
趋势位置强度（0-100）
|
||
"""
|
||
try:
|
||
current_price = self.data.iloc[idx]['high' if fractal_type == 'top' else 'low']
|
||
highest = self.data.iloc[idx]['highest']
|
||
lowest = self.data.iloc[idx]['lowest']
|
||
trend_slope = self.data.iloc[idx]['trend_slope']
|
||
|
||
# 计算在价格区间中的相对位置
|
||
price_range = highest - lowest
|
||
if price_range > 0:
|
||
if fractal_type == 'top':
|
||
# 顶分型：越接近高点越强
|
||
position_ratio = (current_price - lowest) / price_range
|
||
else:
|
||
# 底分型：越接近低点越强
|
||
position_ratio = (highest - current_price) / price_range
|
||
else:
|
||
position_ratio = 0.5
|
||
|
||
position_score = position_ratio * 60 # 位置得分最高60分
|
||
|
||
# 趋势方向得分
|
||
trend_score = 0
|
||
if trend_slope is not None and not np.isnan(trend_slope):
|
||
if fractal_type == 'top' and trend_slope < 0:
|
||
# 顶分型形成在下降趋势中更有效
|
||
trend_score = min(abs(trend_slope) * 1000, 40)
|
||
elif fractal_type == 'bottom' and trend_slope > 0:
|
||
# 底分型形成在上升趋势中更有效
|
||
trend_score = min(abs(trend_slope) * 1000, 40)
|
||
else:
|
||
trend_score = 20 # 趋势方向不匹配给予中等分数
|
||
|
||
total_score = position_score + trend_score
|
||
return max(0, min(total_score, 100))
|
||
|
||
except Exception as e:
|
||
logger.warning(f"计算趋势位置强度失败: {e}")
|
||
return 50 # 默认值
|
||
|
||
def _calculate_enhanced_strength(self, idx: int, fractal_type: str, basic_strength: int) -> Dict[str, float]:
|
||
"""
|
||
计算增强强度评分
|
||
|
||
Args:
|
||
idx: 分型位置
|
||
fractal_type: 分型类型
|
||
basic_strength: 基础强度
|
||
|
||
Returns:
|
||
包含各维度强度的字典
|
||
"""
|
||
# 计算各维度强度
|
||
price_dominance = self._calculate_price_dominance(idx, fractal_type, basic_strength)
|
||
volume_strength = self._calculate_volume_strength(idx, fractal_type, basic_strength)
|
||
trend_position = self._calculate_trend_position_strength(idx, fractal_type)
|
||
|
||
# 基础强度转换为评分（强度越高，分数越高）
|
||
basic_score = min(basic_strength * 20, 60) # 基础强度最高60分
|
||
|
||
# 计算综合强度（加权平均）
|
||
enhanced_strength = (
|
||
basic_score * 0.3 + # 基础强度 30%
|
||
price_dominance * 0.4 + # 价格优势 40%
|
||
volume_strength * 0.2 + # 成交量强度 20%
|
||
trend_position * 0.1 # 趋势位置 10%
|
||
)
|
||
|
||
return {
|
||
'enhanced_strength': enhanced_strength,
|
||
'price_dominance': price_dominance,
|
||
'volume_strength': volume_strength,
|
||
'trend_position': trend_position
|
||
}
|
||
|
||
def is_top_fractal(self, idx: int, strength: int = None) -> Tuple[bool, int]:
|
||
"""
|
||
判断指定位置是否为顶分型
|
||
|
||
Args:
|
||
idx: 检查的位置索引
|
||
strength: 检查强度，如果为None则使用类默认值
|
||
|
||
Returns:
|
||
(是否为顶分型, 实际强度)
|
||
"""
|
||
if strength is None:
|
||
strength = self.min_strength
|
||
|
||
data_len = len(self.data)
|
||
|
||
# 检查边界
|
||
if idx < strength or idx >= data_len - strength:
|
||
return False, 0
|
||
|
||
center_high = self.data.iloc[idx]['high']
|
||
|
||
# 检查左侧K线
|
||
left_valid = True
|
||
for i in range(1, strength + 1):
|
||
if self.data.iloc[idx - i]['high'] >= center_high:
|
||
left_valid = False
|
||
break
|
||
|
||
# 检查右侧K线
|
||
right_valid = True
|
||
for i in range(1, strength + 1):
|
||
if self.data.iloc[idx + i]['high'] >= center_high:
|
||
right_valid = False
|
||
break
|
||
|
||
is_fractal = left_valid and right_valid
|
||
actual_strength = strength if is_fractal else 0
|
||
|
||
return is_fractal, actual_strength
|
||
|
||
def is_bottom_fractal(self, idx: int, strength: int = None) -> Tuple[bool, int]:
|
||
"""
|
||
判断指定位置是否为底分型
|
||
|
||
Args:
|
||
idx: 检查的位置索引
|
||
strength: 检查强度，如果为None则使用类默认值
|
||
|
||
Returns:
|
||
(是否为底分型, 实际强度)
|
||
"""
|
||
if strength is None:
|
||
strength = self.min_strength
|
||
|
||
data_len = len(self.data)
|
||
|
||
# 检查边界
|
||
if idx < strength or idx >= data_len - strength:
|
||
return False, 0
|
||
|
||
center_low = self.data.iloc[idx]['low']
|
||
|
||
# 检查左侧K线
|
||
left_valid = True
|
||
for i in range(1, strength + 1):
|
||
if self.data.iloc[idx - i]['low'] <= center_low:
|
||
left_valid = False
|
||
break
|
||
|
||
# 检查右侧K线
|
||
right_valid = True
|
||
for i in range(1, strength + 1):
|
||
if self.data.iloc[idx + i]['low'] <= center_low:
|
||
right_valid = False
|
||
break
|
||
|
||
is_fractal = left_valid and right_valid
|
||
actual_strength = strength if is_fractal else 0
|
||
|
||
return is_fractal, actual_strength
|
||
|
||
def find_max_strength_fractal(self, idx: int, fractal_type: str, max_strength: int = 5) -> Tuple[bool, int]:
|
||
"""
|
||
寻找指定位置的最大强度分型
|
||
|
||
Args:
|
||
idx: 检查位置
|
||
fractal_type: 'top' 或 'bottom'
|
||
max_strength: 最大检查强度
|
||
|
||
Returns:
|
||
(是否为分型, 最大强度)
|
||
"""
|
||
max_valid_strength = 0
|
||
|
||
for strength in range(self.min_strength, max_strength + 1):
|
||
if fractal_type == 'top':
|
||
is_valid, _ = self.is_top_fractal(idx, strength)
|
||
else:
|
||
is_valid, _ = self.is_bottom_fractal(idx, strength)
|
||
|
||
if is_valid:
|
||
max_valid_strength = strength
|
||
else:
|
||
break # 一旦失败就停止，因为更高强度也不会成功
|
||
|
||
return max_valid_strength > 0, max_valid_strength
|
||
|
||
def detect_fractals(self, use_max_strength: bool = True) -> List[FractalPoint]:
|
||
"""
|
||
检测所有分型
|
||
|
||
Args:
|
||
use_max_strength: 是否使用最大强度检测
|
||
|
||
Returns:
|
||
所有分型点列表
|
||
"""
|
||
fractals = []
|
||
data_len = len(self.data)
|
||
|
||
logger.info(f"开始检测分型，数据长度: {data_len}")
|
||
|
||
# 遍历所有可能的分型位置
|
||
for idx in range(self.min_strength, data_len - self.min_strength):
|
||
timestamp = self.data.index[idx]
|
||
|
||
# 检测顶分型
|
||
if use_max_strength:
|
||
is_top, top_strength = self.find_max_strength_fractal(idx, 'top')
|
||
else:
|
||
is_top, top_strength = self.is_top_fractal(idx)
|
||
|
||
if is_top:
|
||
# 计算增强强度
|
||
strength_metrics = self._calculate_enhanced_strength(idx, 'top', top_strength)
|
||
|
||
fractal = FractalPoint(
|
||
index=idx,
|
||
timestamp=timestamp,
|
||
price=self.data.iloc[idx]['high'],
|
||
fractal_type='top',
|
||
strength=top_strength,
|
||
enhanced_strength=strength_metrics['enhanced_strength'],
|
||
price_dominance=strength_metrics['price_dominance'],
|
||
volume_strength=strength_metrics['volume_strength'],
|
||
trend_position=strength_metrics['trend_position'],
|
||
confirmed=True # 简化处理，认为都已确认
|
||
)
|
||
fractals.append(fractal)
|
||
self.top_fractals.append(fractal)
|
||
|
||
# 检测底分型
|
||
if use_max_strength:
|
||
is_bottom, bottom_strength = self.find_max_strength_fractal(idx, 'bottom')
|
||
else:
|
||
is_bottom, bottom_strength = self.is_bottom_fractal(idx)
|
||
|
||
if is_bottom:
|
||
# 计算增强强度
|
||
strength_metrics = self._calculate_enhanced_strength(idx, 'bottom', bottom_strength)
|
||
|
||
fractal = FractalPoint(
|
||
index=idx,
|
||
timestamp=timestamp,
|
||
price=self.data.iloc[idx]['low'],
|
||
fractal_type='bottom',
|
||
strength=bottom_strength,
|
||
enhanced_strength=strength_metrics['enhanced_strength'],
|
||
price_dominance=strength_metrics['price_dominance'],
|
||
volume_strength=strength_metrics['volume_strength'],
|
||
trend_position=strength_metrics['trend_position'],
|
||
confirmed=True
|
||
)
|
||
fractals.append(fractal)
|
||
self.bottom_fractals.append(fractal)
|
||
|
||
# 按时间排序
|
||
fractals.sort(key=lambda x: x.index)
|
||
self.all_fractals = fractals
|
||
|
||
logger.info(f"检测完成：顶分型 {len(self.top_fractals)} 个，底分型 {len(self.bottom_fractals)} 个")
|
||
return fractals
|
||
|
||
def filter_fractals_by_strength(self, min_strength: int) -> List[FractalPoint]:
|
||
"""
|
||
按强度过滤分型
|
||
|
||
Args:
|
||
min_strength: 最小强度要求
|
||
|
||
Returns:
|
||
过滤后的分型列表
|
||
"""
|
||
return [f for f in self.all_fractals if f.strength >= min_strength]
|
||
|
||
def get_fractal_sequence(self) -> List[FractalPoint]:
|
||
"""
|
||
获取交替的分型序列（顶-底-顶-底...）
|
||
|
||
Returns:
|
||
交替分型序列
|
||
"""
|
||
if not self.all_fractals:
|
||
return []
|
||
|
||
sequence = []
|
||
last_type = None
|
||
|
||
for fractal in self.all_fractals:
|
||
if fractal.fractal_type != last_type:
|
||
sequence.append(fractal)
|
||
last_type = fractal.fractal_type
|
||
|
||
return sequence
|
||
|
||
def validate_fractal_sequence(self, sequence: List[FractalPoint]) -> bool:
|
||
"""
|
||
验证分型序列的有效性
|
||
|
||
Args:
|
||
sequence: 分型序列
|
||
|
||
Returns:
|
||
是否有效
|
||
"""
|
||
if len(sequence) < 2:
|
||
return True
|
||
|
||
for i in range(1, len(sequence)):
|
||
prev_fractal = sequence[i-1]
|
||
curr_fractal = sequence[i]
|
||
|
||
# 检查类型是否交替
|
||
if prev_fractal.fractal_type == curr_fractal.fractal_type:
|
||
return False
|
||
|
||
# 检查价格关系是否合理
|
||
if prev_fractal.fractal_type == 'top':
|
||
# 顶分型后应该是底分型，且价格应该更低
|
||
if curr_fractal.price >= prev_fractal.price:
|
||
return False
|
||
else:
|
||
# 底分型后应该是顶分型，且价格应该更高
|
||
if curr_fractal.price <= prev_fractal.price:
|
||
return False
|
||
|
||
return True
|
||
|
||
def get_fractal_statistics(self) -> Dict:
|
||
"""
|
||
获取分型统计信息（增强版）
|
||
|
||
Returns:
|
||
统计信息字典
|
||
"""
|
||
if not self.all_fractals:
|
||
return {}
|
||
|
||
top_count = len(self.top_fractals)
|
||
bottom_count = len(self.bottom_fractals)
|
||
|
||
# 基础统计
|
||
top_strengths = [f.strength for f in self.top_fractals]
|
||
bottom_strengths = [f.strength for f in self.bottom_fractals]
|
||
|
||
# 增强强度统计
|
||
enhanced_strengths = [f.enhanced_strength for f in self.all_fractals]
|
||
price_dominances = [f.price_dominance for f in self.all_fractals]
|
||
volume_strengths = [f.volume_strength for f in self.all_fractals]
|
||
trend_positions = [f.trend_position for f in self.all_fractals]
|
||
|
||
# 分级统计（按增强强度）
|
||
strong_fractals = [f for f in self.all_fractals if f.enhanced_strength >= 70]
|
||
medium_fractals = [f for f in self.all_fractals if 40 <= f.enhanced_strength < 70]
|
||
weak_fractals = [f for f in self.all_fractals if f.enhanced_strength < 40]
|
||
|
||
stats = {
|
||
'basic_info': {
|
||
'total_fractals': len(self.all_fractals),
|
||
'top_fractals': top_count,
|
||
'bottom_fractals': bottom_count,
|
||
'avg_basic_strength': np.mean([f.strength for f in self.all_fractals]),
|
||
'max_basic_strength': max([f.strength for f in self.all_fractals]),
|
||
},
|
||
'enhanced_strength': {
|
||
'avg_enhanced_strength': np.mean(enhanced_strengths),
|
||
'max_enhanced_strength': max(enhanced_strengths),
|
||
'min_enhanced_strength': min(enhanced_strengths),
|
||
'strong_count': len(strong_fractals), # 强势分型数量
|
||
'medium_count': len(medium_fractals), # 中等分型数量
|
||
'weak_count': len(weak_fractals), # 弱势分型数量
|
||
},
|
||
'dimension_analysis': {
|
||
'avg_price_dominance': np.mean(price_dominances),
|
||
'avg_volume_strength': np.mean(volume_strengths),
|
||
'avg_trend_position': np.mean(trend_positions),
|
||
},
|
||
'top_fractals_detail': {
|
||
'count': top_count,
|
||
'avg_basic_strength': np.mean(top_strengths) if top_strengths else 0,
|
||
'avg_enhanced_strength': np.mean([f.enhanced_strength for f in self.top_fractals]) if self.top_fractals else 0,
|
||
'strong_tops': len([f for f in self.top_fractals if f.enhanced_strength >= 70]),
|
||
},
|
||
'bottom_fractals_detail': {
|
||
'count': bottom_count,
|
||
'avg_basic_strength': np.mean(bottom_strengths) if bottom_strengths else 0,
|
||
'avg_enhanced_strength': np.mean([f.enhanced_strength for f in self.bottom_fractals]) if self.bottom_fractals else 0,
|
||
'strong_bottoms': len([f for f in self.bottom_fractals if f.enhanced_strength >= 70]),
|
||
}
|
||
}
|
||
|
||
return stats
|
||
|
||
def to_dataframe(self) -> pd.DataFrame:
|
||
"""
|
||
将分型转换为DataFrame
|
||
|
||
Returns:
|
||
包含分型信息的DataFrame
|
||
"""
|
||
if not self.all_fractals:
|
||
return pd.DataFrame()
|
||
|
||
data = []
|
||
for fractal in self.all_fractals:
|
||
data.append({
|
||
'timestamp': fractal.timestamp,
|
||
'index': fractal.index,
|
||
'price': fractal.price,
|
||
'type': fractal.fractal_type,
|
||
'strength': fractal.strength,
|
||
'enhanced_strength': fractal.enhanced_strength,
|
||
'price_dominance': fractal.price_dominance,
|
||
'volume_strength': fractal.volume_strength,
|
||
'trend_position': fractal.trend_position,
|
||
'confirmed': fractal.confirmed
|
||
})
|
||
|
||
df = pd.DataFrame(data)
|
||
df.set_index('timestamp', inplace=True)
|
||
return df
|
||
|
||
def update_fractal_confirmation(self, current_idx: int):
|
||
"""
|
||
更新分型确认状态
|
||
|
||
Args:
|
||
current_idx: 当前处理到的K线位置
|
||
"""
|
||
for fractal in self.all_fractals:
|
||
if not fractal.confirmed:
|
||
# 检查是否已经过了足够的确认期
|
||
required_confirmation = fractal.strength
|
||
time_passed = current_idx - fractal.index
|
||
|
||
if time_passed >= required_confirmation:
|
||
fractal.confirmed = True
|
||
|
||
def get_confirmed_fractals(self, current_idx: int) -> List[FractalPoint]:
|
||
"""
|
||
获取已确认的分型列表
|
||
|
||
Args:
|
||
current_idx: 当前处理到的K线位置
|
||
|
||
Returns:
|
||
已确认的分型列表
|
||
"""
|
||
confirmed_fractals = []
|
||
for fractal in self.all_fractals:
|
||
required_confirmation = fractal.strength
|
||
time_passed = current_idx - fractal.index
|
||
|
||
if time_passed >= required_confirmation:
|
||
confirmed_fractals.append(fractal)
|
||
|
||
return confirmed_fractals
|
||
|
||
def detect_real_time_fractals(self, current_idx: int, lookback_periods: int = 50) -> List[FractalPoint]:
|
||
"""
|
||
实时分型检测（避免使用未来数据）
|
||
|
||
Args:
|
||
current_idx: 当前K线位置
|
||
lookback_periods: 回看周期数
|
||
|
||
Returns:
|
||
实时可用的分型列表
|
||
"""
|
||
real_time_fractals = []
|
||
data_len = len(self.data)
|
||
|
||
# 只检测到当前位置之前的分型
|
||
end_idx = min(current_idx, data_len - self.min_strength)
|
||
start_idx = max(self.min_strength, end_idx - lookback_periods)
|
||
|
||
for idx in range(start_idx, end_idx):
|
||
timestamp = self.data.index[idx]
|
||
|
||
# 检测顶分型（但只能检测已确认的）
|
||
is_top, top_strength = self.find_max_strength_fractal(idx, 'top')
|
||
if is_top and current_idx - idx >= top_strength: # 已确认
|
||
strength_metrics = self._calculate_enhanced_strength(idx, 'top', top_strength)
|
||
|
||
fractal = FractalPoint(
|
||
index=idx,
|
||
timestamp=timestamp,
|
||
price=self.data.iloc[idx]['high'],
|
||
fractal_type='top',
|
||
strength=top_strength,
|
||
enhanced_strength=strength_metrics['enhanced_strength'],
|
||
price_dominance=strength_metrics['price_dominance'],
|
||
volume_strength=strength_metrics['volume_strength'],
|
||
trend_position=strength_metrics['trend_position'],
|
||
confirmed=True
|
||
)
|
||
real_time_fractals.append(fractal)
|
||
|
||
# 检测底分型（但只能检测已确认的）
|
||
is_bottom, bottom_strength = self.find_max_strength_fractal(idx, 'bottom')
|
||
if is_bottom and current_idx - idx >= bottom_strength: # 已确认
|
||
strength_metrics = self._calculate_enhanced_strength(idx, 'bottom', bottom_strength)
|
||
|
||
fractal = FractalPoint(
|
||
index=idx,
|
||
timestamp=timestamp,
|
||
price=self.data.iloc[idx]['low'],
|
||
fractal_type='bottom',
|
||
strength=bottom_strength,
|
||
enhanced_strength=strength_metrics['enhanced_strength'],
|
||
price_dominance=strength_metrics['price_dominance'],
|
||
volume_strength=strength_metrics['volume_strength'],
|
||
trend_position=strength_metrics['trend_position'],
|
||
confirmed=True
|
||
)
|
||
real_time_fractals.append(fractal)
|
||
|
||
return real_time_fractals |