kline_app_advanced.py

import dearpygui.dearpygui as dpg
import random
from datetime import datetime, timedelta
import statistics


# 全局变量
kline_data = None
timeframe = "1D"
view_mode = "K线图"


def generate_random_kline_data(num_points=200, timeframe="1D"):
    """生成随机K线数据（A股风格：涨红跌绿）"""
    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 calculate_ma(data, period):
    """计算移动平均线"""
    closes = [d["close"] for d in data]
    ma = []
    for i in range(len(closes)):
        if i < period - 1:
            ma.append(None)
        else:
            ma.append(statistics.mean(closes[i - period + 1:i + 1]))
    return ma


def calculate_macd(data):
    """计算MACD指标"""
    closes = [d["close"] for d in data]
    ema12 = []
    ema26 = []
    macd = []
    signal = []
    histogram = []
    
    # 计算EMA12和EMA26
    for i, close in enumerate(closes):
        if i == 0:
            ema12.append(close)
            ema26.append(close)
        else:
            ema12.append(ema12[-1] * 11/13 + close * 2/13)
            ema26.append(ema26[-1] * 25/27 + close * 2/27)
    
    # 计算MACD线
    for e12, e26 in zip(ema12, ema26):
        macd.append(e12 - e26)
    
    # 计算信号线
    for i, m in enumerate(macd):
        if i == 0:
            signal.append(m)
        else:
            signal.append(signal[-1] * 8/10 + m * 2/10)
    
    # 计算直方图
    for m, s in zip(macd, signal):
        histogram.append(m - s)
    
    return macd, signal, histogram


def create_kline_chart():
    """创建K线图模式"""
    if kline_data is None:
        return
    
    try:
        # 清空现有图表
        for plot_name in ["main_plot", "volume_plot", "indicator_plot"]:
            if dpg.does_item_exist(f"{plot_name}_series"):
                dpg.delete_item(f"{plot_name}_series")
            for item_name in ["candles", "ma5_line", "ma10_line", "ma20_line", "volumes", "macd_line", "signal_line", "macd_hist"]:
                if dpg.does_item_exist(item_name):
                    dpg.delete_item(item_name)
        
        # 提取K线数据
        dates = list(range(len(kline_data)))
        opens = [d["open"] for d in kline_data]
        highs = [d["high"] for d in kline_data]
        lows = [d["low"] for d in kline_data]
        closes = [d["close"] for d in kline_data]
        volumes = [d["volume"] for d in kline_data]
        
        # 计算指标
        ma5 = calculate_ma(kline_data, 5)
        ma10 = calculate_ma(kline_data, 10)
        ma20 = calculate_ma(kline_data, 20)
        macd_line, signal_line, macd_hist = calculate_macd(kline_data)
        
        # 创建蜡烛图数据序列（主图）
        dpg.add_candle_series(
            dates=dates,
            opens=opens,
            closes=closes,
            highs=highs,
            lows=lows,
            tag="candles",
            parent="y_axis",
            # A股风格：涨红跌绿
            bull_color=(255, 0, 0, 255),  # 上涨颜色：红色
            bear_color=(0, 255, 0, 255),  # 下跌颜色：绿色
            weight=0.3
        )
        
        # 创建MA均线（主图）
        dpg.add_line_series(
            dates, ma5, tag="ma5_line", parent="y_axis",
            color=(255, 215, 0, 255), weight=1.5, label="MA5"
        )
        dpg.add_line_series(
            dates, ma10, tag="ma10_line", parent="y_axis",
            color=(0, 191, 255, 255), weight=1.5, label="MA10"
        )
        dpg.add_line_series(
            dates, ma20, tag="ma20_line", parent="y_axis",
            color=(255, 182, 193, 255), weight=1.5, label="MA20"
        )
        
        # 创建成交量数据序列（中间图）
        dpg.add_bar_series(
            x=dates, y=volumes, tag="volumes", parent="volume_axis",
            weight=0.3
        )
        
        # 创建MACD指标（下方图）
        dpg.add_line_series(
            dates, macd_line, tag="macd_line", parent="indicator_axis",
            color=(255, 215, 0, 255), weight=1.0, label="MACD"
        )
        dpg.add_line_series(
            dates, signal_line, tag="signal_line", parent="indicator_axis",
            color=(0, 191, 255, 255), weight=1.0, label="Signal"
        )
        dpg.add_bar_series(
            x=dates, y=macd_hist, tag="macd_hist", parent="indicator_axis",
            weight=0.3
        )
        
        return True
    except Exception as e:
        print(f"创建K线图错误: {e}")
        import traceback
        print(traceback.format_exc())
        return False


def create_time_line_chart():
    """创建分时图模式"""
    if kline_data is None:
        return
    
    try:
        # 清空现有图表
        for plot_name in ["main_plot", "volume_plot", "indicator_plot"]:
            if dpg.does_item_exist(f"{plot_name}_series"):
                dpg.delete_item(f"{plot_name}_series")
            for item_name in ["candles", "ma5_line", "ma10_line", "ma20_line", "volumes", "macd_line", "signal_line", "macd_hist"]:
                if dpg.does_item_exist(item_name):
                    dpg.delete_item(item_name)
        
        # 提取K线数据
        dates = list(range(len(kline_data)))
        closes = [d["close"] for d in kline_data]
        volumes = [d["volume"] for d in kline_data]
        
        # 计算均价线（分时图均价）
        cumulative_price_volume = []
        cumulative_volume = []
        avg_price = []
        
        total_pv = 0
        total_v = 0
        
        for i, (c, v) in enumerate(zip(closes, volumes)):
            total_pv += c * v
            total_v += v
            cumulative_price_volume.append(total_pv)
            cumulative_volume.append(total_v)
            
            if total_v > 0:
                avg_price.append(total_pv / total_v)
            else:
                avg_price.append(c)
        
        # 创建分时线（主图）
        dpg.add_line_series(
            dates, closes, tag="candles", parent="y_axis",
            color=(0, 191, 255, 255), weight=2.0, label="分时线"
        )
        
        # 创建均价线（主图）
        dpg.add_line_series(
            dates, avg_price, tag="ma5_line", parent="y_axis",
            color=(255, 215, 0, 255), weight=1.5, label="均价线"
        )
        
        # 创建成交量数据序列（中间图）
        dpg.add_bar_series(
            x=dates, y=volumes, tag="volumes", parent="volume_axis",
            weight=0.3
        )
        
        return True
    except Exception as e:
        print(f"创建分时图错误: {e}")
        import traceback
        print(traceback.format_exc())
        return False


def regenerate_data():
    """重新生成随机数据"""
    global kline_data
    kline_data = generate_random_kline_data(timeframe=timeframe)
    
    # 根据当前模式重新创建图表
    if view_mode == "K线图":
        create_kline_chart()
    else:
        create_time_line_chart()


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


def change_view_mode(sender, app_data):
    """改变视图模式"""
    global view_mode
    view_mode = app_data
    if view_mode == "K线图":
        create_kline_chart()
    else:
        create_time_line_chart()


def create_main_window():
    """创建主窗口"""
    with dpg.window(label="K线图分析系统", tag="main_window", width=1400, height=900):
        
        # 顶部工具栏
        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):
            # 使用子图布局实现三个固定区域
            with dpg.subplots(3, 1, width=-1, height=-1, tag="kline_plot", row_ratios=[0.6, 0.2, 0.2]):
                
                # 上方子图：K线蜡烛图
                with dpg.plot(width=-1, height=-1, tag="main_plot"):
                    with dpg.plot_axis(dpg.mvXAxis, label="时间", tag="x_axis"):
                        pass
                    with dpg.plot_axis(dpg.mvYAxis, label="价格", tag="y_axis"):
                        pass
                
                # 中间子图：成交量柱状图
                with dpg.plot(width=-1, height=-1, tag="volume_plot"):
                    with dpg.plot_axis(dpg.mvXAxis, label="时间", tag="volume_x_axis"):
                        pass
                    with dpg.plot_axis(dpg.mvYAxis, label="成交量", tag="volume_axis"):
                        pass
                
                # 下方子图：技术指标
                with dpg.plot(width=-1, height=-1, tag="indicator_plot"):
                    with dpg.plot_axis(dpg.mvXAxis, label="时间", tag="indicator_x_axis"):
                        pass
                    with dpg.plot_axis(dpg.mvYAxis, label="MACD", tag="indicator_axis", opposite=True):
                        pass
                
                # 添加图例
                dpg.add_plot_legend(parent="kline_plot")


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()
        
        # 配置图表主题
        with dpg.theme(tag="plot_theme"):
            with dpg.theme_component(dpg.mvPlot):
                dpg.add_theme_color(dpg.mvPlotCol_PlotBg, (30, 30, 30, 255))
                dpg.add_theme_color(dpg.mvPlotCol_PlotBorder, (60, 60, 60, 255))
                dpg.add_theme_color(dpg.mvPlotCol_AxisGrid, (50, 50, 50, 200))
                dpg.add_theme_color(dpg.mvPlotCol_AxisText, (200, 200, 200, 255))
                dpg.add_theme_color(dpg.mvPlotCol_TitleText, (255, 255, 255, 255))
        
        dpg.bind_item_theme("kline_plot", "plot_theme")
        
        # 生成初始数据
        generate_random_kline_data(timeframe=timeframe)
        
        # 创建图表
        create_kline_chart()
        
        # 创建视图
        dpg.create_viewport(title="K线图分析系统", width=1400, height=900)
        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())
    finally:
        try:
            dpg.destroy_context()
        except:
            pass


if __name__ == "__main__":
    main()