kline_app_matplotlib.py

import dearpygui.dearpygui as dpg
import numpy as np
from datetime import datetime, timedelta
import random
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
from matplotlib.patches import Rectangle


# 全局变量
kline_data = None
current_timeframe = "1D"  # 默认时间周期
view_mode = "K线图"  # 默认视图模式


def generate_random_kline_data(num_points=200, timeframe="1D"):
    """生成随机K线数据"""
    global kline_data
    now = datetime.now()
    data = []
    
    # 根据时间周期计算时间间隔
    if timeframe == "1D":
        delta = timedelta(days=1)
    elif timeframe == "4H":
        delta = timedelta(hours=4)
    elif timeframe == "1H":
        delta = timedelta(hours=1)
    elif timeframe == "15M":
        delta = timedelta(minutes=15)
    elif timeframe == "5M":
        delta = timedelta(minutes=5)
    elif timeframe == "1M":
        delta = timedelta(minutes=1)
    else:
        delta = timedelta(days=1)
    
    # 初始价格
    open_price = 100.0
    for i in range(num_points):
        timestamp = now - delta * (num_points - i)
        
        # 随机价格波动
        volatility = 0.02
        high = open_price * (1 + random.uniform(0, volatility))
        low = open_price * (1 - random.uniform(0, volatility))
        close = open_price + random.uniform(-volatility * open_price, volatility * open_price)
        
        # 随机成交量
        volume = random.randint(1000, 10000)
        
        data.append({
            "timestamp": timestamp,
            "open": open_price,
            "high": high,
            "low": low,
            "close": close,
            "volume": volume
        })
        
        open_price = close
    
    kline_data = data
    return data


def draw_kline_chart():
    """使用matplotlib绘制K线图"""
    if kline_data is None:
        return None
    
    # 创建图形
    fig, ax = plt.subplots(figsize=(10, 6), facecolor='#1a1a1a')
    ax.set_facecolor('#1a1a1a')
    
    # 绘制K线
    for i, data in enumerate(kline_data):
        timestamp = i
        open_p = data["open"]
        high_p = data["high"]
        low_p = data["low"]
        close_p = data["close"]
        
        # 确定K线颜色（涨/跌）
        color = '#00ff00' if close_p >= open_p else '#ff0000'
        
        # 绘制K线实体
        rect = Rectangle((timestamp - 0.3, min(open_p, close_p)),
                        0.6, max(open_p, close_p) - min(open_p, close_p),
                        facecolor=color, edgecolor=color)
        ax.add_patch(rect)
        
        # 绘制影线
        plt.plot([timestamp, timestamp], [high_p, low_p], color=color, linewidth=1.5)
    
    # 设置坐标轴
    ax.set_xlim(-1, len(kline_data))
    ax.set_ylim(min(d["low"] for d in kline_data) * 0.98, max(d["high"] for d in kline_data) * 1.02)
    
    # 设置X轴标签
    ax.set_xlabel("时间", color='#ffffff')
    ax.set_ylabel("价格", color='#ffffff')
    ax.set_title(f"{current_timeframe} K线图", color='#ffffff')
    
    # 配置坐标轴刻度
    ax.tick_params(axis='x', colors='#ffffff')
    ax.tick_params(axis='y', colors='#ffffff')
    
    # 配置网格
    ax.grid(True, color='#444444', linestyle='--', alpha=0.3)
    
    # 移除顶部和右侧边框
    for spine in ax.spines.values():
        spine.set_edgecolor('#444444')
    
    # 保存为PNG图像
    plt.tight_layout()
    plt.savefig("kline_chart.png", facecolor='#1a1a1a', transparent=True)
    plt.close()
    
    return "kline_chart.png"


def draw_time_line_chart():
    """使用matplotlib绘制分时图"""
    if kline_data is None:
        return None
    
    # 创建图形
    fig, ax = plt.subplots(figsize=(10, 6), facecolor='#1a1a1a')
    ax.set_facecolor('#1a1a1a')
    
    # 绘制价格走势
    times = list(range(len(kline_data)))
    closes = [d["close"] for d in kline_data]
    
    ax.plot(times, closes, color='#0099ff', linewidth=2)
    
    # 填充区域
    ax.fill_between(times, closes, color='#0099ff', alpha=0.2)
    
    # 设置坐标轴
    ax.set_xlim(0, len(kline_data) - 1)
    ax.set_ylim(min(closes) * 0.98, max(closes) * 1.02)
    
    # 设置X轴标签
    ax.set_xlabel("时间", color='#ffffff')
    ax.set_ylabel("价格", color='#ffffff')
    ax.set_title(f"{current_timeframe} 分时图", color='#ffffff')
    
    # 配置坐标轴刻度
    ax.tick_params(axis='x', colors='#ffffff')
    ax.tick_params(axis='y', colors='#ffffff')
    
    # 配置网格
    ax.grid(True, color='#444444', linestyle='--', alpha=0.3)
    
    # 移除顶部和右侧边框
    for spine in ax.spines.values():
        spine.set_edgecolor('#444444')
    
    # 保存为PNG图像
    plt.tight_layout()
    plt.savefig("time_line_chart.png", facecolor='#1a1a1a', transparent=True)
    plt.close()
    
    return "time_line_chart.png"


def update_kline_display():
    """更新K线图显示"""
    if kline_data is None:
        return
    
    # 绘制K线图
    chart_path = draw_kline_chart()
    
    # 显示图像
    if chart_path:
        # 加载新图像
        width, height, channels, data = dpg.load_image(chart_path)
        
        # 更新现有的纹理
        dpg.configure_item("kline_image", width=width, height=height, default_value=data)


def update_time_line_chart():
    """更新分时图显示"""
    if kline_data is None:
        return
    
    # 绘制分时图
    chart_path = draw_time_line_chart()
    
    # 显示图像
    if chart_path:
        # 加载新图像
        width, height, channels, data = dpg.load_image(chart_path)
        
        # 更新现有的纹理
        dpg.configure_item("time_line_image", width=width, height=height, default_value=data)


def regenerate_data():
    """重新生成随机数据"""
    generate_random_kline_data(timeframe=current_timeframe)
    if view_mode == "K线图":
        update_kline_display()
    else:
        update_time_line_chart()


def change_timeframe(sender, app_data):
    """改变时间周期"""
    global current_timeframe
    current_timeframe = app_data
    regenerate_data()


def change_view_mode(sender, app_data):
    """改变视图模式"""
    global view_mode
    view_mode = app_data
    if view_mode == "K线图":
        update_kline_display()
        dpg.show_item("kline_display")
        dpg.hide_item("time_line_display")
    else:
        update_time_line_chart()
        dpg.hide_item("kline_display")
        dpg.show_item("time_line_display")


def create_main_window():
    """创建主窗口"""
    # 加载图像纹理
    with dpg.texture_registry():
        dpg.add_static_texture(1, 1, [0, 0, 0], tag="kline_image")
        dpg.add_static_texture(1, 1, [0, 0, 0], tag="time_line_image")
    
    with dpg.window(label="K线图分析系统", tag="main_window", width=1200, height=800):
        
        # 顶部工具栏
        with dpg.group(horizontal=True):
            dpg.add_text("时间周期:")
            dpg.add_radio_button(
                ["1M", "5M", "15M", "1H", "4H", "1D"],
                label="时间周期",
                default_value="1D",
                callback=change_timeframe,
                tag="timeframe_radio"
            )
            
            dpg.add_spacer(width=20)
            
            dpg.add_text("视图模式:")
            dpg.add_radio_button(
                ["K线图", "分时图"],
                label="视图模式",
                default_value="K线图",
                callback=change_view_mode,
                tag="view_mode_radio"
            )
            
            dpg.add_spacer(width=20)
            
            dpg.add_button(
                label="重新生成数据",
                callback=regenerate_data,
                tag="regenerate_btn",
                width=150
            )
        
        dpg.add_separator()
        
        # 主图区域
        with dpg.child_window(height=-1, width=-1, border=True):
            # K线图显示区域
            with dpg.group(tag="kline_display"):
                dpg.add_image("kline_image", tag="kline_display_image", width=1100, height=600)
            
            # 分时图显示区域（初始隐藏）
            with dpg.group(tag="time_line_display"):
                dpg.add_image("time_line_image", tag="time_line_display_image", width=1100, height=600)
            dpg.hide_item("time_line_display")


def create_sidebar():
    """创建侧边栏"""
    with dpg.window(label="指标", tag="indicator_window", width=250, height=800, pos=(1200, 0)):
        
        dpg.add_text("技术指标")
        dpg.add_separator()
        
        with dpg.collapsing_header(label="MA均线"):
            dpg.add_checkbox(label="MA5", default_value=True)
            dpg.add_checkbox(label="MA10", default_value=True)
            dpg.add_checkbox(label="MA20", default_value=True)
            dpg.add_checkbox(label="MA60", default_value=True)
        
        with dpg.collapsing_header(label="MACD"):
            dpg.add_checkbox(label="MACD", default_value=True)
        
        with dpg.collapsing_header(label="KDJ"):
            dpg.add_checkbox(label="KDJ", default_value=True)
        
        with dpg.collapsing_header(label="RSI"):
            dpg.add_checkbox(label="RSI", default_value=True)
        
        dpg.add_separator()
        
        dpg.add_text("数据统计")
        if kline_data:
            dpg.add_text(f"最高价格: {max(d['high'] for d in kline_data):.2f}")
            dpg.add_text(f"最低价格: {min(d['low'] for d in kline_data):.2f}")
            dpg.add_text(f"平均价格: {sum(d['close'] for d in kline_data)/len(kline_data):.2f}")
            dpg.add_text(f"总成交量: {sum(d['volume'] for d in kline_data):,}")


def main():
    """主函数"""
    try:
        # 初始化Dear PyGui
        dpg.create_context()
        
        # 配置主题
        with dpg.theme(tag="theme"):
            with dpg.theme_component(dpg.mvAll):
                dpg.add_theme_color(dpg.mvThemeCol_WindowBg, (30, 30, 30, 255))
                dpg.add_theme_color(dpg.mvThemeCol_ChildBg, (40, 40, 40, 255))
                dpg.add_theme_color(dpg.mvThemeCol_Text, (255, 255, 255, 255))
                dpg.add_theme_style(dpg.mvStyleVar_WindowPadding, 10, 10)
                dpg.add_theme_style(dpg.mvStyleVar_FramePadding, 5, 5)
        
        dpg.bind_theme("theme")
        
        # 创建主窗口和侧边栏
        create_main_window()
        create_sidebar()
        
        # 生成初始数据
        generate_random_kline_data()
        
        # 更新图表显示
        update_kline_display()
        
        # 创建视图
        dpg.create_viewport(title="K线图分析系统", width=1450, height=800)
        dpg.setup_dearpygui()
        
        # 显示窗口
        dpg.show_viewport()
        dpg.start_dearpygui()
        dpg.destroy_context()
    except Exception as e:
        print(f"程序启动错误: {e}")
        import traceback
        print("完整错误信息:")
        print(traceback.format_exc())


if __name__ == "__main__":
    main()