test_kline_functions.py

import random
from datetime import datetime, timedelta
import statistics


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


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


# 测试数据生成
print("1. 测试数据生成...")
data = generate_random_kline_data(num_points=10, timeframe="1D")
print(f"生成的数据长度: {len(data)}")
print("数据示例:")
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("2. 测试移动平均线计算...")
ma5 = calculate_ma(data, 5)
ma10 = calculate_ma(data, 10)
ma20 = calculate_ma(data, 20)

print(f"MA5 计算结果长度: {len(ma5)}")
print(f"MA10 计算结果长度: {len(ma10)}")
print(f"MA20 计算结果长度: {len(ma20)}")
print(f"MA5 前3个值: {[f'{x:.2f}' if x else 'None' for x in ma5[:3]]}")
print(f"MA10 前3个值: {[f'{x:.2f}' if x else 'None' for x in ma10[:3]]}")
print(f"MA20 前3个值: {[f'{x:.2f}' if x else 'None' for x in ma20[:3]]}")
print()


# 测试MACD指标计算
print("3. 测试MACD指标计算...")
macd_line, signal_line, macd_hist = calculate_macd(data)

print(f"MACD线 计算结果长度: {len(macd_line)}")
print(f"信号线 计算结果长度: {len(signal_line)}")
print(f"直方图 计算结果长度: {len(macd_hist)}")
print(f"MACD线 前3个值: {[f'{x:.4f}' for x in macd_line[:3]]}")
print(f"信号线 前3个值: {[f'{x:.4f}' for x in signal_line[:3]]}")
print(f"直方图 前3个值: {[f'{x:.4f}' for x in macd_hist[:3]]}")
print()


# 测试数据提取
print("4. 测试数据提取...")
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()


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