crypto_quant/test_ma_methods.py

"""
均线多空判定方法测试脚本

本脚本用于测试和比较不同均线多空判定方法的有效性。
"""

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from core.db.db_market_data import DBMarketData
from core.biz.metrics_calculation import MetricsCalculation
import logging
from config import OKX_MONITOR_CONFIG, MYSQL_CONFIG, WINDOW_SIZE
# plt支持中文
plt.rcParams['font.family'] = ['SimHei']

# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def get_real_data(symbol, bar, start, end):
    mysql_user = MYSQL_CONFIG.get("user", "xch")
    mysql_password = MYSQL_CONFIG.get("password", "")
    if not mysql_password:
        raise ValueError("MySQL password is not set")
    mysql_host = MYSQL_CONFIG.get("host", "localhost")
    mysql_port = MYSQL_CONFIG.get("port", 3306)
    mysql_database = MYSQL_CONFIG.get("database", "okx")

    db_url = f"mysql+pymysql://{mysql_user}:{mysql_password}@{mysql_host}:{mysql_port}/{mysql_database}"
    db_market_data = DBMarketData(db_url)
    data = db_market_data.query_market_data_by_symbol_bar(
            symbol, bar, start, end
        )
    if data is None:
        logging.warning(
            f"获取行情数据失败: {symbol} {bar} 从 {start} 到 {end}"
        )
        return None
    else:
        if len(data) == 0:
            logging.warning(
                f"获取行情数据为空: {symbol} {bar} 从 {start} 到 {end}"
            )
            return None
        else:
            if isinstance(data, list):
                data = pd.DataFrame(data)
            elif isinstance(data, dict):
                data = pd.DataFrame([data])
    return data

def generate_test_data(n=1000):
    """生成测试数据"""
    np.random.seed(42)
    
    # 生成价格数据
    price = 100
    prices = []
    for i in range(n):
        # 添加趋势和随机波动
        trend = np.sin(i * 0.1) * 2  # 周期性趋势
        noise = np.random.normal(0, 1)  # 随机噪声
        price += trend + noise
        prices.append(price)
    
    # 创建DataFrame
    data = pd.DataFrame({
        'timestamp': pd.date_range('2023-01-01', periods=n, freq='H'),
        'close': prices,
        'open': [p * (1 + np.random.normal(0, 0.01)) for p in prices],
        'high': [p * (1 + abs(np.random.normal(0, 0.02))) for p in prices],
        'low': [p * (1 - abs(np.random.normal(0, 0.02))) for p in prices],
        'volume': np.random.randint(1000, 10000, n)
    })
    
    return data

def test_ma_methods():
    """测试不同的均线多空判定方法"""
    print("=== 均线多空判定方法测试 ===\n")
    
    # 生成测试数据
    data = get_real_data("BTC-USDT", "15m", "2025-07-01 00:00:00", "2025-08-07 00:00:00")
    # data = generate_test_data(1000)
    print(f"生成测试数据: {len(data)} 条记录")
    
    # 初始化指标计算器
    metrics = MetricsCalculation()
    
    # 计算均线
    data = metrics.ma5102030(data)
    data = metrics.calculate_ma_price_percent(data)
    
    # 测试不同方法
    methods = [
        "weighted_voting",
        "trend_strength", 
        "ma_alignment",
        "statistical",
        "hybrid"
    ]
    
    results = {}
    
    for method in methods:
        print(f"\n--- 测试方法: {method} ---")
        
        # 复制数据避免相互影响
        test_data = data.copy()
        
        # 应用方法
        test_data = metrics.set_ma_long_short_advanced(test_data, method=method)
        
        # 统计结果
        long_count = (test_data['ma_long_short'] == '多').sum()
        short_count = (test_data['ma_long_short'] == '空').sum()
        neutral_count = (test_data['ma_long_short'] == '震荡').sum()
        
        results[method] = {
            'long': long_count,
            'short': short_count, 
            'neutral': neutral_count,
            'data': test_data
        }
        
        print(f"多头信号: {long_count} ({long_count/len(test_data)*100:.1f}%)")
        print(f"空头信号: {short_count} ({short_count/len(test_data)*100:.1f}%)")
        print(f"震荡信号: {neutral_count} ({neutral_count/len(test_data)*100:.1f}%)")
    
    return results, data

def compare_methods(results, original_data):
    """比较不同方法的结果"""
    print("\n=== 方法比较分析 ===\n")
    
    # 创建比较表格
    comparison_data = []
    for method, result in results.items():
        total = result['long'] + result['short'] + result['neutral']
        comparison_data.append({
            '方法': method,
            '多头信号': f"{result['long']} ({result['long']/total*100:.1f}%)",
            '空头信号': f"{result['short']} ({result['short']/total*100:.1f}%)",
            '震荡信号': f"{result['neutral']} ({result['neutral']/total*100:.1f}%)",
            '信号总数': result['long'] + result['short']
        })
    
    comparison_df = pd.DataFrame(comparison_data)
    print("信号分布比较:")
    print(comparison_df.to_string(index=False))
    
    # 分析信号一致性
    print("\n信号一致性分析:")
    methods = list(results.keys())
    
    for i in range(len(methods)):
        for j in range(i+1, len(methods)):
            method1, method2 = methods[i], methods[j]
            data1 = results[method1]['data']['ma_long_short']
            data2 = results[method2]['data']['ma_long_short']
            
            # 计算一致性
            agreement = (data1 == data2).sum()
            agreement_rate = agreement / len(data1) * 100
            
            print(f"{method1} vs {method2}: {agreement_rate:.1f}% 一致")

def visualize_results(results, original_data):
    """可视化结果"""
    print("\n=== 生成可视化图表 ===")
    
    # 选择前2000个数据点进行可视化
    plot_data = original_data.head(1000).copy()
    
    # 添加不同方法的结果
    for method, result in results.items():
        plot_data[f'{method}_signal'] = result['data'].head(1000)['ma_long_short']
    
    # 创建图表
    fig, axes = plt.subplots(4, 2, figsize=(15, 16))
    fig.suptitle('均线多空判定方法比较', fontsize=16)
    
    # 价格和均线图
    ax1 = axes[0, 0]
    ax1.plot(plot_data.index, plot_data['close'], label='价格', alpha=0.7)
    ax1.plot(plot_data.index, plot_data['ma5'], label='MA5', alpha=0.8)
    ax1.plot(plot_data.index, plot_data['ma10'], label='MA10', alpha=0.8)
    ax1.plot(plot_data.index, plot_data['ma20'], label='MA20', alpha=0.8)
    ax1.plot(plot_data.index, plot_data['ma30'], label='MA30', alpha=0.8)
    ax1.set_title('价格和均线')
    ax1.legend()
    ax1.grid(True, alpha=0.3)
    
    # 均线偏离度图
    ax2 = axes[0, 1]
    ax2.plot(plot_data.index, plot_data['ma5_close_diff'], label='MA5偏离度', alpha=0.7)
    ax2.plot(plot_data.index, plot_data['ma10_close_diff'], label='MA10偏离度', alpha=0.7)
    ax2.plot(plot_data.index, plot_data['ma20_close_diff'], label='MA20偏离度', alpha=0.7)
    ax2.plot(plot_data.index, plot_data['ma30_close_diff'], label='MA30偏离度', alpha=0.7)
    ax2.axhline(y=0, color='black', linestyle='--', alpha=0.5)
    ax2.set_title('均线偏离度')
    ax2.legend()
    ax2.grid(True, alpha=0.3)
    
    # 不同方法的信号图
    methods = list(results.keys())
    colors = ['red', 'blue', 'green', 'orange', 'purple']
    
    for i, method in enumerate(methods):
        print(f"绘制{method}方法信号")
        row = 1 + (i // 2)  # 从第2行开始，每行2个图
        col = i % 2         # 0或1
        ax = axes[row, col]
        
        # 绘制信号
        long_signals = plot_data[plot_data[f'{method}_signal'] == '多'].index
        short_signals = plot_data[plot_data[f'{method}_signal'] == '空'].index
        
        ax.plot(plot_data.index, plot_data['close'], color='gray', alpha=0.5, label='价格')
        ax.scatter(long_signals, plot_data.loc[long_signals, 'close'], 
                  color='red', marker='^', s=50, label='多头信号', alpha=0.8)
        ax.scatter(short_signals, plot_data.loc[short_signals, 'close'], 
                  color='blue', marker='v', s=50, label='空头信号', alpha=0.8)
        
        ax.set_title(f'{method}方法信号')
        ax.legend()
        ax.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.savefig('ma_methods_comparison.png', dpi=300, bbox_inches='tight')
    print("图表已保存为: ma_methods_comparison.png")
    plt.show()

def main():
    """主函数"""
    print("开始测试均线多空判定方法...")
    
    # 测试方法
    results, original_data = test_ma_methods()
    
    # 比较结果
    compare_methods(results, original_data)
    
    # 可视化结果
    try:
        visualize_results(results, original_data)
    except Exception as e:
        print(f"可视化失败: {e}")
    
    print("\n=== 测试完成 ===")
    print("\n建议:")
    print("1. 加权投票方法适合大多数市场环境")
    print("2. 趋势强度方法适合趋势明显的市场")
    print("3. 均线排列方法适合技术分析")
    print("4. 统计方法适合波动较大的市场")
    print("5. 混合方法综合了多种优势，但计算复杂度较高")

if __name__ == "__main__":
    main()