test_kline_core.py

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


def generate_random_kline_data(num_points=200, timeframe="1D"):
    """生成随机K线数据（A股风格：涨红跌绿）"""
    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
    
    return data


# 测试数据生成
print("测试数据生成...")
data = generate_random_kline_data(num_points=10, timeframe="1D")

print("生成的数据长度:", len(data))
print("\n数据示例:")
for i, d in enumerate(data[:3]):
    print(f"第{i+1}条 K 线:")
    print(f"时间: {d['timestamp']}")
    print(f"开盘价: {d['open']:.2f}")
    print(f"收盘价: {d['close']:.2f}")
    print(f"最高价: {d['high']:.2f}")
    print(f"最低价: {d['low']:.2f}")
    print(f"成交量: {d['volume']}")
    print(f"涨跌: {'涨' if d['close'] >= d['open'] else '跌'}")
    print()


# 测试数据提取
print("测试数据提取...")
dates = list(range(len(data)))
opens = [d["open"] for d in data]
highs = [d["high"] for d in data]
lows = [d["low"] for d in data]
closes = [d["close"] for d in data]
volumes = [d["volume"] for d in data]

print(f"日期数组长度: {len(dates)}")
print(f"开盘价数组长度: {len(opens)}")
print(f"成交量数组长度: {len(volumes)}")

print("\n所有测试通过！")