#!/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)