421 lines
14 KiB
Python
421 lines
14 KiB
Python
"""
|
|
缠论综合分析器：整合所有核心模块进行完整的缠论分析
|
|
"""
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
from typing import Dict, List, Optional, Tuple
|
|
import logging
|
|
|
|
from .kline import KLine
|
|
from .fractal import Fractal, FractalPoint
|
|
from .stroke import Stroke, StrokeElement
|
|
from .segment import Segment, SegmentElement
|
|
from .central_bank import CentralBank, CentralBankElement
|
|
from .trading_signal import TradingSignal, TradingPoint
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ChanAnalyzer:
|
|
"""缠论综合分析器"""
|
|
|
|
def __init__(self, kline_data: pd.DataFrame):
|
|
"""
|
|
初始化缠论分析器
|
|
|
|
Args:
|
|
kline_data: 原始K线数据
|
|
"""
|
|
self.original_data = kline_data.copy()
|
|
self.processed_data = None
|
|
|
|
# 各模块实例
|
|
self.kline_processor = None
|
|
self.fractal_detector = None
|
|
self.stroke_detector = None
|
|
self.segment_detector = None
|
|
self.central_bank_detector = None
|
|
self.trading_signal_detector = None
|
|
|
|
# 分析结果
|
|
self.fractals = []
|
|
self.strokes = []
|
|
self.segments = []
|
|
self.central_banks = []
|
|
self.trading_points = []
|
|
|
|
# 分析状态
|
|
self.is_analyzed = False
|
|
|
|
def run_full_analysis(self, fractal_strength: int = 1) -> Dict:
|
|
"""
|
|
运行完整的缠论分析
|
|
|
|
Args:
|
|
fractal_strength: 分型强度要求
|
|
|
|
Returns:
|
|
分析结果摘要
|
|
"""
|
|
logger.info("开始完整缠论分析...")
|
|
|
|
try:
|
|
# 1. 处理K线包含关系
|
|
self._process_klines()
|
|
|
|
# 2. 识别分型
|
|
self._detect_fractals(fractal_strength)
|
|
|
|
# 3. 生成笔
|
|
self._detect_strokes()
|
|
|
|
# 4. 生成线段
|
|
self._detect_segments()
|
|
|
|
# 5. 识别中枢
|
|
self._detect_central_banks()
|
|
|
|
# 6. 识别买卖点
|
|
self._detect_trading_signals()
|
|
|
|
self.is_analyzed = True
|
|
|
|
# 生成分析摘要
|
|
summary = self._generate_analysis_summary()
|
|
|
|
logger.info("缠论分析完成")
|
|
return summary
|
|
|
|
except Exception as e:
|
|
logger.error(f"缠论分析失败: {e}")
|
|
raise
|
|
|
|
def _process_klines(self):
|
|
"""处理K线包含关系"""
|
|
logger.info("处理K线包含关系...")
|
|
|
|
self.kline_processor = KLine(self.original_data)
|
|
self.processed_data = self.kline_processor.get_processed_data()
|
|
|
|
logger.info(f"K线处理完成：{len(self.original_data)} -> {len(self.processed_data)}")
|
|
|
|
def _detect_fractals(self, strength: int = 1):
|
|
"""识别分型"""
|
|
logger.info("识别分型...")
|
|
|
|
self.fractal_detector = Fractal(self.processed_data, min_strength=strength)
|
|
self.fractals = self.fractal_detector.detect_fractals()
|
|
|
|
logger.info(f"分型识别完成：共 {len(self.fractals)} 个")
|
|
|
|
def _detect_strokes(self):
|
|
"""生成笔"""
|
|
logger.info("生成笔...")
|
|
|
|
if not self.fractals:
|
|
logger.warning("没有分型，无法生成笔")
|
|
return
|
|
|
|
self.stroke_detector = Stroke(self.fractals, self.processed_data)
|
|
self.strokes = self.stroke_detector.detect_strokes()
|
|
|
|
logger.info(f"笔生成完成：共 {len(self.strokes)} 个")
|
|
|
|
def _detect_segments(self):
|
|
"""生成线段"""
|
|
logger.info("生成线段...")
|
|
|
|
if not self.strokes:
|
|
logger.warning("没有笔，无法生成线段")
|
|
return
|
|
|
|
self.segment_detector = Segment(self.strokes)
|
|
self.segments = self.segment_detector.detect_segments()
|
|
|
|
logger.info(f"线段生成完成：共 {len(self.segments)} 个")
|
|
|
|
def _detect_central_banks(self):
|
|
"""识别中枢"""
|
|
logger.info("识别中枢...")
|
|
|
|
if not self.segments:
|
|
logger.warning("没有线段，无法识别中枢")
|
|
return
|
|
|
|
self.central_bank_detector = CentralBank(self.segments)
|
|
self.central_banks = self.central_bank_detector.detect_central_banks()
|
|
|
|
logger.info(f"中枢识别完成：共 {len(self.central_banks)} 个")
|
|
|
|
def _detect_trading_signals(self):
|
|
"""识别买卖点"""
|
|
logger.info("识别买卖点...")
|
|
|
|
self.trading_signal_detector = TradingSignal(
|
|
self.central_banks, self.segments, self.strokes
|
|
)
|
|
|
|
# 检测中枢相关的买卖点
|
|
central_bank_signals = []
|
|
if self.central_banks:
|
|
central_bank_signals = self.trading_signal_detector.detect_all_trading_points()
|
|
|
|
# 如果中枢数量不足，增加基于分型的买卖点识别
|
|
fractal_signals = []
|
|
if len(self.central_banks) < 2 and self.fractals:
|
|
logger.info("中枢数量不足，启用分型买卖点识别")
|
|
fractal_signals = self.trading_signal_detector.detect_fractal_based_signals(
|
|
self.fractals, self.processed_data
|
|
)
|
|
|
|
# 合并所有信号
|
|
all_signals = central_bank_signals + fractal_signals
|
|
|
|
# 去重和排序
|
|
unique_signals = []
|
|
seen_keys = set()
|
|
for signal in sorted(all_signals, key=lambda x: x.timestamp):
|
|
# 创建唯一键：时间+类型+类别
|
|
key = f"{signal.timestamp.strftime('%Y%m%d%H%M')}_{signal.signal_type}_{signal.point_class}"
|
|
if key not in seen_keys:
|
|
unique_signals.append(signal)
|
|
seen_keys.add(key)
|
|
|
|
self.trading_points = unique_signals
|
|
|
|
# 统计信号
|
|
central_count = len(central_bank_signals)
|
|
fractal_count = len(fractal_signals)
|
|
total_count = len(unique_signals)
|
|
|
|
class_counts = {'first': 0, 'second': 0, 'third': 0}
|
|
for signal in unique_signals:
|
|
class_counts[signal.point_class] += 1
|
|
|
|
logger.info(f"买卖点识别完成：中枢相关 {central_count} 个，分型相关 {fractal_count} 个，"
|
|
f"一类 {class_counts['first']} 个，二类 {class_counts['second']} 个，"
|
|
f"三类 {class_counts['third']} 个，总计 {total_count} 个")
|
|
|
|
def _generate_analysis_summary(self) -> Dict:
|
|
"""生成分析摘要"""
|
|
summary = {
|
|
'data_info': {
|
|
'original_klines': len(self.original_data),
|
|
'processed_klines': len(self.processed_data) if self.processed_data is not None else 0,
|
|
'date_range': {
|
|
'start': self.original_data.index.min(),
|
|
'end': self.original_data.index.max()
|
|
}
|
|
},
|
|
'fractal_info': {
|
|
'total': len(self.fractals),
|
|
'top': len([f for f in self.fractals if f.fractal_type == 'top']),
|
|
'bottom': len([f for f in self.fractals if f.fractal_type == 'bottom'])
|
|
},
|
|
'stroke_info': {
|
|
'total': len(self.strokes),
|
|
'up': len([s for s in self.strokes if s.direction == 1]),
|
|
'down': len([s for s in self.strokes if s.direction == -1])
|
|
},
|
|
'segment_info': {
|
|
'total': len(self.segments),
|
|
'up': len([s for s in self.segments if s.direction == 1]),
|
|
'down': len([s for s in self.segments if s.direction == -1])
|
|
},
|
|
'central_bank_info': {
|
|
'total': len(self.central_banks),
|
|
'levels': self._get_central_bank_levels()
|
|
},
|
|
'trading_signal_info': {
|
|
'total': len(self.trading_points),
|
|
'buy_points': len([p for p in self.trading_points if p.signal_type == 'buy']),
|
|
'sell_points': len([p for p in self.trading_points if p.signal_type == 'sell']),
|
|
'by_class': self._get_signal_class_distribution()
|
|
}
|
|
}
|
|
|
|
return summary
|
|
|
|
def _get_central_bank_levels(self) -> Dict:
|
|
"""获取中枢级别分布"""
|
|
levels = {}
|
|
for cb in self.central_banks:
|
|
levels[cb.level] = levels.get(cb.level, 0) + 1
|
|
return levels
|
|
|
|
def _get_signal_class_distribution(self) -> Dict:
|
|
"""获取买卖点类别分布"""
|
|
distribution = {}
|
|
for point in self.trading_points:
|
|
key = f"{point.point_class}_class"
|
|
distribution[key] = distribution.get(key, 0) + 1
|
|
return distribution
|
|
|
|
def get_latest_signals(self, hours: int = 24) -> List[TradingPoint]:
|
|
"""
|
|
获取最近的买卖点信号
|
|
|
|
Args:
|
|
hours: 最近多少小时
|
|
|
|
Returns:
|
|
最近的信号列表
|
|
"""
|
|
if not self.is_analyzed or not self.trading_points:
|
|
return []
|
|
|
|
latest_time = self.processed_data.index[-1]
|
|
cutoff_time = latest_time - pd.Timedelta(hours=hours)
|
|
|
|
return [p for p in self.trading_points if p.timestamp >= cutoff_time]
|
|
|
|
def get_current_market_structure(self) -> Dict:
|
|
"""
|
|
获取当前市场结构
|
|
|
|
Returns:
|
|
当前市场结构信息
|
|
"""
|
|
if not self.is_analyzed:
|
|
return {}
|
|
|
|
current_price = self.processed_data['close'].iloc[-1]
|
|
current_time = self.processed_data.index[-1]
|
|
|
|
# 最近的中枢
|
|
recent_central_banks = [cb for cb in self.central_banks
|
|
if (current_time - cb.end_time).total_seconds() < 7*24*3600] # 7天内
|
|
|
|
# 最近的线段趋势
|
|
recent_segments = [seg for seg in self.segments
|
|
if (current_time - seg.end_time).total_seconds() < 3*24*3600] # 3天内
|
|
|
|
# 当前趋势方向
|
|
current_trend = self._determine_current_trend(recent_segments)
|
|
|
|
# 支撑阻力位
|
|
support_resistance = self._calculate_support_resistance(recent_central_banks)
|
|
|
|
return {
|
|
'current_price': current_price,
|
|
'current_time': current_time,
|
|
'trend': current_trend,
|
|
'recent_central_banks': len(recent_central_banks),
|
|
'support_resistance': support_resistance,
|
|
'market_phase': self._determine_market_phase()
|
|
}
|
|
|
|
def _determine_current_trend(self, recent_segments: List[SegmentElement]) -> str:
|
|
"""确定当前趋势"""
|
|
if not recent_segments:
|
|
return "unclear"
|
|
|
|
# 按时间排序取最近3个线段
|
|
recent_segments = sorted(recent_segments, key=lambda x: x.end_time)[-3:]
|
|
|
|
up_count = sum(1 for seg in recent_segments if seg.direction == 1)
|
|
down_count = sum(1 for seg in recent_segments if seg.direction == -1)
|
|
|
|
if up_count > down_count:
|
|
return "upward"
|
|
elif down_count > up_count:
|
|
return "downward"
|
|
else:
|
|
return "sideways"
|
|
|
|
def _calculate_support_resistance(self, central_banks: List[CentralBankElement]) -> Dict:
|
|
"""计算支撑阻力位"""
|
|
if not central_banks:
|
|
return {}
|
|
|
|
# 按强度排序，取最强的几个中枢
|
|
strong_cbs = sorted(central_banks, key=lambda x: x.strength, reverse=True)[:3]
|
|
|
|
supports = []
|
|
resistances = []
|
|
|
|
for cb in strong_cbs:
|
|
supports.append(cb.low_price)
|
|
resistances.append(cb.high_price)
|
|
|
|
return {
|
|
'support_levels': sorted(supports),
|
|
'resistance_levels': sorted(resistances, reverse=True)
|
|
}
|
|
|
|
def _determine_market_phase(self) -> str:
|
|
"""确定市场阶段"""
|
|
if not self.central_banks:
|
|
return "unknown"
|
|
|
|
# 最近的中枢
|
|
latest_cb = self.central_banks[-1] if self.central_banks else None
|
|
|
|
if not latest_cb:
|
|
return "trending"
|
|
|
|
current_time = self.processed_data.index[-1]
|
|
time_since_cb = (current_time - latest_cb.end_time).total_seconds() / 3600 # 小时
|
|
|
|
if time_since_cb < 24:
|
|
return "consolidation" # 盘整
|
|
else:
|
|
return "trending" # 趋势
|
|
|
|
def export_results(self) -> Dict[str, pd.DataFrame]:
|
|
"""
|
|
导出所有分析结果
|
|
|
|
Returns:
|
|
包含各种分析结果的DataFrame字典
|
|
"""
|
|
if not self.is_analyzed:
|
|
raise ValueError("尚未进行分析，请先调用 run_full_analysis()")
|
|
|
|
results = {}
|
|
|
|
# 处理后的K线数据
|
|
if self.processed_data is not None:
|
|
results['klines'] = self.processed_data
|
|
|
|
# 分型数据
|
|
if self.fractal_detector:
|
|
results['fractals'] = self.fractal_detector.to_dataframe()
|
|
|
|
# 笔数据
|
|
if self.stroke_detector:
|
|
results['strokes'] = self.stroke_detector.to_dataframe()
|
|
|
|
# 线段数据
|
|
if self.segment_detector:
|
|
results['segments'] = self.segment_detector.to_dataframe()
|
|
|
|
# 中枢数据
|
|
if self.central_bank_detector:
|
|
results['central_banks'] = self.central_bank_detector.to_dataframe()
|
|
|
|
# 买卖点数据
|
|
if self.trading_signal_detector:
|
|
results['trading_signals'] = self.trading_signal_detector.to_dataframe()
|
|
|
|
return results
|
|
|
|
def get_visualization_data(self) -> Dict:
|
|
"""
|
|
获取可视化所需的数据
|
|
|
|
Returns:
|
|
可视化数据字典
|
|
"""
|
|
if not self.is_analyzed:
|
|
return {}
|
|
|
|
return {
|
|
'klines': self.processed_data,
|
|
'fractals': self.fractals,
|
|
'strokes': self.strokes,
|
|
'segments': self.segments,
|
|
'central_banks': self.central_banks,
|
|
'trading_points': self.trading_points
|
|
} |