445 lines
15 KiB
Python
445 lines
15 KiB
Python
"""
|
|
缠论可视化模块：使用Plotly生成交互式图表
|
|
"""
|
|
|
|
import plotly.graph_objects as go
|
|
import plotly.express as px
|
|
from plotly.subplots import make_subplots
|
|
import pandas as pd
|
|
import numpy as np
|
|
from typing import Dict, List, Optional
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ChanVisualizer:
|
|
"""缠论可视化器"""
|
|
|
|
def __init__(self):
|
|
"""初始化可视化器"""
|
|
self.colors = {
|
|
'up_candle': '#26a69a',
|
|
'down_candle': '#ef5350',
|
|
'fractal_top': '#ff6b6b',
|
|
'fractal_bottom': '#4ecdc4',
|
|
'stroke_up': '#2e86de',
|
|
'stroke_down': '#f39c12',
|
|
'segment_up': '#0984e3',
|
|
'segment_down': '#e17055',
|
|
'central_bank': 'rgba(155, 89, 182, 0.3)',
|
|
'buy_signal': '#00b894',
|
|
'sell_signal': '#d63031'
|
|
}
|
|
|
|
def create_comprehensive_chart(self, data: Dict) -> go.Figure:
|
|
"""
|
|
创建综合缠论分析图表
|
|
|
|
Args:
|
|
data: 包含所有分析数据的字典
|
|
|
|
Returns:
|
|
Plotly图表对象
|
|
"""
|
|
if not data or 'klines' not in data:
|
|
return go.Figure()
|
|
|
|
# 创建子图
|
|
fig = make_subplots(
|
|
rows=2, cols=1,
|
|
shared_xaxes=True,
|
|
vertical_spacing=0.1,
|
|
subplot_titles=('缠论分析图', '成交量'),
|
|
row_heights=[0.8, 0.2]
|
|
)
|
|
|
|
# 添加K线图
|
|
self._add_candlestick(fig, data['klines'])
|
|
|
|
# 添加分型
|
|
if 'fractals' in data and data['fractals']:
|
|
self._add_fractals(fig, data['fractals'])
|
|
|
|
# 添加笔
|
|
if 'strokes' in data and data['strokes']:
|
|
self._add_strokes(fig, data['strokes'])
|
|
|
|
# 添加线段
|
|
if 'segments' in data and data['segments']:
|
|
self._add_segments(fig, data['segments'])
|
|
|
|
# 添加中枢
|
|
if 'central_banks' in data and data['central_banks']:
|
|
self._add_central_banks(fig, data['central_banks'])
|
|
|
|
# 添加买卖点
|
|
if 'trading_points' in data and data['trading_points']:
|
|
self._add_trading_signals(fig, data['trading_points'])
|
|
|
|
# 添加成交量
|
|
self._add_volume(fig, data['klines'])
|
|
|
|
# 更新布局
|
|
self._update_layout(fig)
|
|
|
|
return fig
|
|
|
|
def _add_candlestick(self, fig: go.Figure, klines: pd.DataFrame):
|
|
"""添加K线图"""
|
|
fig.add_trace(
|
|
go.Candlestick(
|
|
x=klines.index,
|
|
open=klines['open'],
|
|
high=klines['high'],
|
|
low=klines['low'],
|
|
close=klines['close'],
|
|
name='K线',
|
|
increasing=dict(line=dict(color=self.colors['up_candle'])),
|
|
decreasing=dict(line=dict(color=self.colors['down_candle']))
|
|
),
|
|
row=1, col=1
|
|
)
|
|
|
|
def _add_fractals(self, fig: go.Figure, fractals: List):
|
|
"""添加分型标记"""
|
|
top_fractals = [f for f in fractals if f.fractal_type == 'top']
|
|
bottom_fractals = [f for f in fractals if f.fractal_type == 'bottom']
|
|
|
|
if top_fractals:
|
|
fig.add_trace(
|
|
go.Scatter(
|
|
x=[f.timestamp for f in top_fractals],
|
|
y=[f.price for f in top_fractals],
|
|
mode='markers',
|
|
marker=dict(
|
|
symbol='triangle-down',
|
|
size=8,
|
|
color=self.colors['fractal_top']
|
|
),
|
|
name='顶分型',
|
|
hovertemplate='顶分型<br>时间: %{x}<br>价格: %{y}<br>强度: %{customdata}<extra></extra>',
|
|
customdata=[f.strength for f in top_fractals]
|
|
),
|
|
row=1, col=1
|
|
)
|
|
|
|
if bottom_fractals:
|
|
fig.add_trace(
|
|
go.Scatter(
|
|
x=[f.timestamp for f in bottom_fractals],
|
|
y=[f.price for f in bottom_fractals],
|
|
mode='markers',
|
|
marker=dict(
|
|
symbol='triangle-up',
|
|
size=8,
|
|
color=self.colors['fractal_bottom']
|
|
),
|
|
name='底分型',
|
|
hovertemplate='底分型<br>时间: %{x}<br>价格: %{y}<br>强度: %{customdata}<extra></extra>',
|
|
customdata=[f.strength for f in bottom_fractals]
|
|
),
|
|
row=1, col=1
|
|
)
|
|
|
|
def _add_strokes(self, fig: go.Figure, strokes: List):
|
|
"""添加笔"""
|
|
for stroke in strokes:
|
|
color = self.colors['stroke_up'] if stroke.direction == 1 else self.colors['stroke_down']
|
|
|
|
fig.add_trace(
|
|
go.Scatter(
|
|
x=[stroke.start_fractal.timestamp, stroke.end_fractal.timestamp],
|
|
y=[stroke.start_fractal.price, stroke.end_fractal.price],
|
|
mode='lines',
|
|
line=dict(color=color, width=2),
|
|
name='笔' if stroke == strokes[0] else '',
|
|
showlegend=stroke == strokes[0],
|
|
hovertemplate=f'笔<br>方向: {"上升" if stroke.direction == 1 else "下降"}<br>长度: {stroke.length:.2f}<br>强度: {stroke.strength:.2f}<extra></extra>'
|
|
),
|
|
row=1, col=1
|
|
)
|
|
|
|
def _add_segments(self, fig: go.Figure, segments: List):
|
|
"""添加线段"""
|
|
for segment in segments:
|
|
color = self.colors['segment_up'] if segment.direction == 1 else self.colors['segment_down']
|
|
|
|
fig.add_trace(
|
|
go.Scatter(
|
|
x=[segment.start_time, segment.end_time],
|
|
y=[segment.start_price, segment.end_price],
|
|
mode='lines',
|
|
line=dict(color=color, width=4, dash='dash'),
|
|
name='线段' if segment == segments[0] else '',
|
|
showlegend=segment == segments[0],
|
|
hovertemplate=f'线段<br>方向: {"上升" if segment.direction == 1 else "下降"}<br>长度: {segment.length:.2f}<br>强度: {segment.strength:.2f}<extra></extra>'
|
|
),
|
|
row=1, col=1
|
|
)
|
|
|
|
def _add_central_banks(self, fig: go.Figure, central_banks: List):
|
|
"""添加中枢"""
|
|
for cb in central_banks:
|
|
# 添加中枢矩形区域
|
|
fig.add_shape(
|
|
type="rect",
|
|
x0=cb.start_time,
|
|
y0=cb.low_price,
|
|
x1=cb.end_time,
|
|
y1=cb.high_price,
|
|
fillcolor=self.colors['central_bank'],
|
|
opacity=0.3,
|
|
line=dict(color="rgba(155, 89, 182, 0.8)", width=1),
|
|
row=1, col=1
|
|
)
|
|
|
|
# 添加中枢标签
|
|
fig.add_annotation(
|
|
x=cb.start_time + (cb.end_time - cb.start_time) / 2,
|
|
y=cb.center_price,
|
|
text=f"中枢({cb.level})",
|
|
showarrow=False,
|
|
font=dict(size=10, color="purple"),
|
|
bgcolor="rgba(255,255,255,0.8)",
|
|
row=1, col=1
|
|
)
|
|
|
|
def _add_trading_signals(self, fig: go.Figure, trading_points: List):
|
|
"""添加买卖点信号"""
|
|
buy_points = [p for p in trading_points if p.signal_type == 'buy']
|
|
sell_points = [p for p in trading_points if p.signal_type == 'sell']
|
|
|
|
if buy_points:
|
|
fig.add_trace(
|
|
go.Scatter(
|
|
x=[p.timestamp for p in buy_points],
|
|
y=[p.price for p in buy_points],
|
|
mode='markers',
|
|
marker=dict(
|
|
symbol='triangle-up',
|
|
size=12,
|
|
color=self.colors['buy_signal']
|
|
),
|
|
name='买点',
|
|
hovertemplate='%{customdata}<br>时间: %{x}<br>价格: %{y}<br>强度: %{text}<extra></extra>',
|
|
customdata=[p.description for p in buy_points],
|
|
text=[f"{p.strength:.3f}" for p in buy_points]
|
|
),
|
|
row=1, col=1
|
|
)
|
|
|
|
if sell_points:
|
|
fig.add_trace(
|
|
go.Scatter(
|
|
x=[p.timestamp for p in sell_points],
|
|
y=[p.price for p in sell_points],
|
|
mode='markers',
|
|
marker=dict(
|
|
symbol='triangle-down',
|
|
size=12,
|
|
color=self.colors['sell_signal']
|
|
),
|
|
name='卖点',
|
|
hovertemplate='%{customdata}<br>时间: %{x}<br>价格: %{y}<br>强度: %{text}<extra></extra>',
|
|
customdata=[p.description for p in sell_points],
|
|
text=[f"{p.strength:.3f}" for p in sell_points]
|
|
),
|
|
row=1, col=1
|
|
)
|
|
|
|
def _add_volume(self, fig: go.Figure, klines: pd.DataFrame):
|
|
"""添加成交量"""
|
|
colors = [self.colors['up_candle'] if close >= open_ else self.colors['down_candle']
|
|
for close, open_ in zip(klines['close'], klines['open'])]
|
|
|
|
fig.add_trace(
|
|
go.Bar(
|
|
x=klines.index,
|
|
y=klines['volume'],
|
|
name='成交量',
|
|
marker_color=colors,
|
|
showlegend=False
|
|
),
|
|
row=2, col=1
|
|
)
|
|
|
|
def _update_layout(self, fig: go.Figure):
|
|
"""更新图表布局"""
|
|
fig.update_layout(
|
|
title=dict(
|
|
text="缠论分析图表",
|
|
x=0.5,
|
|
font=dict(size=20)
|
|
),
|
|
xaxis_rangeslider_visible=False,
|
|
height=800,
|
|
showlegend=True,
|
|
legend=dict(
|
|
orientation="h",
|
|
yanchor="bottom",
|
|
y=1.02,
|
|
xanchor="right",
|
|
x=1
|
|
),
|
|
margin=dict(l=50, r=50, t=100, b=50),
|
|
plot_bgcolor='white',
|
|
paper_bgcolor='white'
|
|
)
|
|
|
|
# 更新X轴
|
|
fig.update_xaxes(
|
|
title_text="时间",
|
|
showgrid=True,
|
|
gridwidth=1,
|
|
gridcolor='lightgray'
|
|
)
|
|
|
|
# 更新Y轴
|
|
fig.update_yaxes(
|
|
title_text="价格",
|
|
showgrid=True,
|
|
gridwidth=1,
|
|
gridcolor='lightgray',
|
|
row=1, col=1
|
|
)
|
|
|
|
fig.update_yaxes(
|
|
title_text="成交量",
|
|
row=2, col=1
|
|
)
|
|
|
|
def create_statistics_charts(self, data: Dict) -> List[go.Figure]:
|
|
"""
|
|
创建统计分析图表
|
|
|
|
Args:
|
|
data: 分析数据
|
|
|
|
Returns:
|
|
统计图表列表
|
|
"""
|
|
charts = []
|
|
|
|
# 分型强度分布
|
|
if 'fractals' in data and data['fractals']:
|
|
fractal_chart = self._create_fractal_strength_chart(data['fractals'])
|
|
charts.append(fractal_chart)
|
|
|
|
# 买卖点类别分布
|
|
if 'trading_points' in data and data['trading_points']:
|
|
signal_chart = self._create_signal_distribution_chart(data['trading_points'])
|
|
charts.append(signal_chart)
|
|
|
|
# 中枢级别分布
|
|
if 'central_banks' in data and data['central_banks']:
|
|
cb_chart = self._create_central_bank_chart(data['central_banks'])
|
|
charts.append(cb_chart)
|
|
|
|
return charts
|
|
|
|
def _create_fractal_strength_chart(self, fractals: List) -> go.Figure:
|
|
"""创建分型强度分布图"""
|
|
strengths = [f.strength for f in fractals]
|
|
types = [f.fractal_type for f in fractals]
|
|
|
|
df = pd.DataFrame({'strength': strengths, 'type': types})
|
|
|
|
fig = px.histogram(
|
|
df,
|
|
x='strength',
|
|
color='type',
|
|
title='分型强度分布',
|
|
labels={'strength': '强度', 'type': '类型'},
|
|
color_discrete_map={'top': self.colors['fractal_top'], 'bottom': self.colors['fractal_bottom']}
|
|
)
|
|
|
|
return fig
|
|
|
|
def _create_signal_distribution_chart(self, trading_points: List) -> go.Figure:
|
|
"""创建买卖点分布图"""
|
|
classes = [f"{p.point_class}类{p.signal_type}点" for p in trading_points]
|
|
|
|
fig = px.pie(
|
|
values=[classes.count(c) for c in set(classes)],
|
|
names=list(set(classes)),
|
|
title='买卖点类别分布'
|
|
)
|
|
|
|
return fig
|
|
|
|
def _create_central_bank_chart(self, central_banks: List) -> go.Figure:
|
|
"""创建中枢分析图"""
|
|
levels = [cb.level for cb in central_banks]
|
|
strengths = [cb.strength for cb in central_banks]
|
|
|
|
fig = go.Figure()
|
|
|
|
for level in set(levels):
|
|
level_strengths = [s for l, s in zip(levels, strengths) if l == level]
|
|
fig.add_trace(go.Box(
|
|
y=level_strengths,
|
|
name=level,
|
|
boxpoints='all'
|
|
))
|
|
|
|
fig.update_layout(
|
|
title='中枢强度分布（按级别）',
|
|
xaxis_title='中枢级别',
|
|
yaxis_title='强度'
|
|
)
|
|
|
|
return fig
|
|
|
|
def create_market_structure_chart(self, market_structure: Dict) -> go.Figure:
|
|
"""
|
|
创建市场结构图
|
|
|
|
Args:
|
|
market_structure: 市场结构数据
|
|
|
|
Returns:
|
|
市场结构图表
|
|
"""
|
|
fig = go.Figure()
|
|
|
|
# 当前价格线
|
|
if 'current_price' in market_structure:
|
|
fig.add_hline(
|
|
y=market_structure['current_price'],
|
|
line_dash="dash",
|
|
line_color="black",
|
|
annotation_text=f"当前价格: {market_structure['current_price']:.2f}"
|
|
)
|
|
|
|
# 支撑阻力位
|
|
if 'support_resistance' in market_structure:
|
|
sr = market_structure['support_resistance']
|
|
|
|
# 支撑位
|
|
if 'support_levels' in sr:
|
|
for i, support in enumerate(sr['support_levels']):
|
|
fig.add_hline(
|
|
y=support,
|
|
line_dash="dot",
|
|
line_color=self.colors['buy_signal'],
|
|
annotation_text=f"支撑{i+1}: {support:.2f}"
|
|
)
|
|
|
|
# 阻力位
|
|
if 'resistance_levels' in sr:
|
|
for i, resistance in enumerate(sr['resistance_levels']):
|
|
fig.add_hline(
|
|
y=resistance,
|
|
line_dash="dot",
|
|
line_color=self.colors['sell_signal'],
|
|
annotation_text=f"阻力{i+1}: {resistance:.2f}"
|
|
)
|
|
|
|
fig.update_layout(
|
|
title="市场结构分析",
|
|
yaxis_title="价格",
|
|
height=400
|
|
)
|
|
|
|
return fig |