cyb50-quant/trend-quality-evaluator/optimize_params_simple.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
趋势质量评估器 - 参数优化测试 (简化版)
测试不同参数组合的回测表现
"""

import sys
sys.path.insert(0, '/root/.openclaw/workspace/trend-quality-evaluator')

import numpy as np
import pandas as pd
from trend_quality_evaluator import fetch_stock_data
import warnings
warnings.filterwarnings('ignore')

print("="*70)
print("趋势质量评估器 - 参数优化测试")
print("="*70)

# 获取数据
df = fetch_stock_data("399673", "2019-01-01", "2026-03-06", "d")
if df is None:
    print("数据获取失败")
    exit(1)

print(f"\n✓ 数据获取成功: {len(df)}条")

# 预计算指标
def calc_indicators(data):
    d = data.copy()
    high, low, close = d['high'], d['low'], d['close']
    
    # ADX
    plus_dm = high.diff().where(high.diff() > 0, 0)
    minus_dm = low.diff().abs().where(low.diff().abs() > 0, 0)
    tr = pd.concat([high-low, (high-close.shift()).abs(), (low-close.shift()).abs()], axis=1).max(axis=1)
    atr = tr.rolling(14).mean()
    dx = (plus_dm.rolling(14).mean() / atr * 100).abs()
    d['adx'] = dx.rolling(14).mean()
    
    # MA斜率
    d['ma20'] = d['close'].rolling(20).mean()
    d['ma_slope'] = d['ma20'] / d['ma20'].shift(5)
    
    # ATR比率
    d['atr_ratio'] = tr.rolling(14).mean() / tr.rolling(50).mean()
    
    # 成交量
    d['vol_ratio'] = d['volume'] / d['volume'].rolling(20).mean()
    
    d['above_ma20'] = d['close'] > d['ma20']
    return d

data = calc_indicators(df)

# 测试的参数组合 (简化版)
test_configs = [
    # 配置1: 原配置
    {'name': '原配置', 'adx_t': 25, 'adx_w': 30, 'ma_t': 1.002, 'ma_w': 25, 'vol_t': 0.8, 'vol_w': 20, 'volu_t': 1.5, 'volu_w': 10},
    # 配置2: 提高ADX权重
    {'name': '高ADX权重', 'adx_t': 25, 'adx_w': 35, 'ma_t': 1.002, 'ma_w': 20, 'vol_t': 0.8, 'vol_w': 20, 'volu_t': 1.5, 'volu_w': 10},
    # 配置3: 降低ADX阈值
    {'name': '低ADX阈值', 'adx_t': 20, 'adx_w': 30, 'ma_t': 1.002, 'ma_w': 25, 'vol_t': 0.8, 'vol_w': 20, 'volu_t': 1.5, 'volu_w': 10},
    # 配置4: 提高MA权重
    {'name': '高MA权重', 'adx_t': 25, 'adx_w': 25, 'ma_t': 1.002, 'ma_w': 30, 'vol_t': 0.8, 'vol_w': 20, 'volu_t': 1.5, 'volu_w': 10},
    # 配置5: 更严格成交量
    {'name': '严格成交量', 'adx_t': 25, 'adx_w': 30, 'ma_t': 1.002, 'ma_w': 25, 'vol_t': 0.8, 'vol_w': 15, 'volu_t': 2.0, 'volu_w': 15},
    # 配置6: 宽松波动率
    {'name': '宽松波动率', 'adx_t': 25, 'adx_w': 30, 'ma_t': 1.002, 'ma_w': 25, 'vol_t': 0.9, 'vol_w': 20, 'volu_t': 1.5, 'volu_w': 10},
    # 配置7: 综合优化1
    {'name': '综合优化1', 'adx_t': 22, 'adx_w': 32, 'ma_t': 1.001, 'ma_w': 28, 'vol_t': 0.85, 'vol_w': 18, 'volu_t': 1.8, 'volu_w': 12},
    # 配置8: 综合优化2
    {'name': '综合优化2', 'adx_t': 20, 'adx_w': 35, 'ma_t': 1.003, 'ma_w': 25, 'vol_t': 0.8, 'vol_w': 15, 'volu_t': 2.0, 'volu_w': 10},
]

results = []

for cfg in test_configs:
    print(f"\n测试: {cfg['name']}...")
    
    scores = []
    for i in range(60, len(data)):
        row = data.iloc[i]
        
        # 计算各因子得分
        adx_s = cfg['adx_w'] if row['adx'] >= cfg['adx_t'] else cfg['adx_w'] * (row['adx'] / cfg['adx_t'])
        
        if row['ma_slope'] >= cfg['ma_t'] * 1.5:
            ma_s = cfg['ma_w']
        elif row['ma_slope'] >= cfg['ma_t']:
            ma_s = cfg['ma_w'] * 0.7
        elif row['ma_slope'] >= 1.0:
            ma_s = cfg['ma_w'] * 0.3
        else:
            ma_s = 0
        
        if row['atr_ratio'] <= cfg['vol_t'] * 0.75:
            vol_s = cfg['vol_w']
        elif row['atr_ratio'] <= cfg['vol_t']:
            vol_s = cfg['vol_w'] * 0.7
        elif row['atr_ratio'] <= 1.0:
            vol_s = cfg['vol_w'] * 0.3
        else:
            vol_s = 0
        
        if row['vol_ratio'] >= cfg['volu_t'] * 1.3:
            volu_s = cfg['volu_w']
        elif row['vol_ratio'] >= cfg['volu_t']:
            volu_s = cfg['volu_w'] * 0.7
        elif row['vol_ratio'] >= 1.0:
            volu_s = cfg['volu_w'] * 0.3
        else:
            volu_s = 0
        
        tf_s = 15 if row['above_ma20'] else 0
        
        total = adx_s + ma_s + vol_s + volu_s + tf_s
        is_trade = total >= 60
        
        scores.append({
            'date': data.index[i],
            'close': row['close'],
            'score': total,
            'is_trade': is_trade,
            'adx_s': adx_s, 'ma_s': ma_s, 'vol_s': vol_s, 'volu_s': volu_s, 'tf_s': tf_s
        })
    
    scores_df = pd.DataFrame(scores).set_index('date')
    scores_df['ret_20d'] = scores_df['close'].pct_change(20).shift(-20) * 100
    
    t_mask = scores_df['is_trade']
    trade_days = t_mask.sum()
    
    if trade_days > 0:
        ret_20d = scores_df[t_mask]['ret_20d'].dropna()
        result = {
            'name': cfg['name'],
            'adx_t': cfg['adx_t'], 'adx_w': cfg['adx_w'],
            'ma_t': cfg['ma_t'], 'ma_w': cfg['ma_w'],
            'vol_t': cfg['vol_t'], 'vol_w': cfg['vol_w'],
            'volu_t': cfg['volu_t'], 'volu_w': cfg['volu_w'],
            'trade_pct': trade_days / len(scores_df) * 100,
            'avg_score': scores_df['score'].mean(),
            'ret_20d': ret_20d.mean() if len(ret_20d) > 0 else 0,
            'win_rate': (ret_20d > 0).mean() * 100 if len(ret_20d) > 0 else 0,
            'sharpe': ret_20d.mean() / ret_20d.std() if len(ret_20d) > 0 and ret_20d.std() > 0 else 0,
        }
        results.append(result)
        print(f"  可交易: {result['trade_pct']:.1f}%, 20日收益: {result['ret_20d']:+.2f}%, 胜率: {result['win_rate']:.1f}%")

# 排序结果
print("\n" + "="*70)
print("参数优化结果排名")
print("="*70)

# 按20日收益排序
results_by_return = sorted(results, key=lambda x: x['ret_20d'], reverse=True)
print("\n【按20日收益排序】")
print(f"{'排名':<4} {'配置名称':<12} {'可交易%':<10} {'20日收益':<12} {'胜率':<10} {'夏普':<8}")
print("-"*70)
for i, r in enumerate(results_by_return[:5]):
    print(f"{i+1:<4} {r['name']:<12} {r['trade_pct']:>8.1f}% {r['ret_20d']:>+10.2f}% {r['win_rate']:>8.1f}% {r['sharpe']:>6.2f}")

# 按胜率排序
results_by_winrate = sorted(results, key=lambda x: x['win_rate'], reverse=True)
print("\n【按胜率排序】")
print(f"{'排名':<4} {'配置名称':<12} {'可交易%':<10} {'20日收益':<12} {'胜率':<10} {'夏普':<8}")
print("-"*70)
for i, r in enumerate(results_by_winrate[:5]):
    print(f"{i+1:<4} {r['name']:<12} {r['trade_pct']:>8.1f}% {r['ret_20d']:>+10.2f}% {r['win_rate']:>8.1f}% {r['sharpe']:>6.2f}")

# 综合最佳
print("\n【综合推荐配置】")
best = results_by_return[0]
print(f"配置名称: {best['name']}")
print(f"ADX阈值: {best['adx_t']}, 权重: {best['adx_w']}")
print(f"MA斜率阈值: {best['ma_t']}, 权重: {best['ma_w']}")
print(f"波动率阈值: {best['vol_t']}, 权重: {best['vol_w']}")
print(f"成交量阈值: {best['volu_t']}, 权重: {best['volu_w']}")
print(f"\n绩效:")
print(f"  可交易比例: {best['trade_pct']:.1f}%")
print(f"  20日平均收益: {best['ret_20d']:+.2f}%")
print(f"  胜率: {best['win_rate']:.1f}%")
print(f"  夏普比率: {best['sharpe']:.2f}")

# 保存结果
results_df = pd.DataFrame(results)
results_df = results_df.sort_values('ret_20d', ascending=False)
results_df.to_csv('/root/.openclaw/workspace/trend-quality-evaluator/param_optimization_results.csv', index=False)
print("\n✓ 结果已保存: param_optimization_results.csv")

print("\n" + "="*70)
print("参数优化测试完成!")
print("="*70)