|
|
| """
|
| 华尔街量化回测集成模块 - TANZG 单股快速回测
|
| """
|
|
|
| import sys, os, warnings
|
| warnings.filterwarnings('ignore')
|
|
|
| _bt_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), 'backtest'))
|
| if _bt_dir not in sys.path:
|
| sys.path.insert(0, _bt_dir)
|
|
|
| import pandas as pd
|
| import numpy as np
|
| from datetime import datetime, timedelta
|
|
|
| from rules import VolumePriceRules, KCandlestickRules, ChipDistributionRules, MoneyFlowRules
|
| from scoring import evaluate_signal
|
| from indicators import calc_all_indicators
|
|
|
|
|
| def quick_backtest(kline_df: pd.DataFrame, lookback_days: int = 250) -> dict:
|
| """
|
| 单股快速回测 - 验证华尔街模型在该股票上的历史表现
|
| 使用最近 lookback_days 天数据，信号触发后持有10天的盈亏统计
|
|
|
| Args:
|
| kline_df: 日线K线DataFrame
|
| lookback_days: 回测使用数据天数（默认250个交易日≈1年）
|
|
|
| Returns:
|
| {
|
| 'total_signals': int, # 总信号次数
|
| 'win_count': int, # 盈利次数
|
| 'lose_count': int, # 亏损次数
|
| 'win_rate': float, # 胜率 %
|
| 'avg_profit_pct': float, # 平均盈利%
|
| 'avg_loss_pct': float, # 平均亏损%
|
| 'profit_loss_ratio': float, # 盈亏比
|
| 'hold_days': int, # 持仓天数
|
| 'conclusion': str, # 回测结论
|
| }
|
| """
|
| if kline_df is None or len(kline_df) < 100:
|
| return {'error': 'K线数据不足，无法回测'}
|
|
|
| df = kline_df.copy()
|
|
|
|
|
| col_map = {}
|
| for c in df.columns:
|
| cl = c.lower().strip()
|
| if cl == 'date': col_map[c] = 'date'
|
| elif cl == 'open': col_map[c] = 'open'
|
| elif cl == 'high': col_map[c] = 'high'
|
| elif cl == 'low': col_map[c] = 'low'
|
| elif cl == 'close': col_map[c] = 'close'
|
| elif cl in ('volume', 'vol'): col_map[c] = 'volume'
|
| elif cl in ('amount',): col_map[c] = 'amount'
|
| if col_map:
|
| df = df.rename(columns=col_map)
|
| for col in ['date', 'open', 'high', 'low', 'close', 'volume']:
|
| if col not in df.columns:
|
| return {'error': f'缺少必要列: {col}'}
|
|
|
| df['date'] = pd.to_datetime(df['date'])
|
| df = df.sort_values('date').reset_index(drop=True)
|
|
|
|
|
| df['fund_net_inflow'] = 0.0
|
| df['fund_net_inflow_ratio'] = 0.0
|
|
|
|
|
| df = calc_all_indicators(df)
|
|
|
|
|
| df = df.tail(lookback_days + 60).reset_index(drop=True)
|
|
|
|
|
| hold_days = 10
|
| signals_found = []
|
| min_idx = 60
|
|
|
| for i in range(min_idx, len(df) - hold_days):
|
| row = df.iloc[i]
|
| history = df.iloc[:i]
|
| if len(history) < 60:
|
| continue
|
|
|
| vp = VolumePriceRules.evaluate_all(row, history)
|
| kc = KCandlestickRules.evaluate_all(row, history)
|
| cp = ChipDistributionRules.evaluate_all(row, history)
|
| mf = MoneyFlowRules.evaluate_all(row, history)
|
| sig = evaluate_signal(vp, kc, cp, mf)
|
|
|
| if sig['signal_level'] in ('A', 'B') and not sig['veto_triggered']:
|
|
|
| buy_price = row['close']
|
| future_idx = i + hold_days
|
| if future_idx < len(df):
|
| sell_price = df.iloc[future_idx]['close']
|
| profit_pct = (sell_price / buy_price - 1) * 100
|
| signals_found.append({
|
| 'date': str(row['date'].date()),
|
| 'score': sig['total_score'],
|
| 'level': sig['signal_level'],
|
| 'buy_price': round(float(buy_price), 2),
|
| 'sell_price': round(float(sell_price), 2),
|
| 'profit_pct': round(float(profit_pct), 2),
|
| })
|
|
|
| if not signals_found:
|
| return {
|
| 'total_signals': 0,
|
| 'win_count': 0,
|
| 'lose_count': 0,
|
| 'win_rate': 0,
|
| 'avg_profit_pct': 0,
|
| 'avg_loss_pct': 0,
|
| 'profit_loss_ratio': 0,
|
| 'hold_days': hold_days,
|
| 'conclusion': '近一年无足够的买入信号样本，无法进行有效回测',
|
| }
|
|
|
| winning = [s for s in signals_found if s['profit_pct'] > 0]
|
| losing = [s for s in signals_found if s['profit_pct'] <= 0]
|
|
|
| win_rate = len(winning) / len(signals_found) * 100 if signals_found else 0
|
| avg_win = np.mean([s['profit_pct'] for s in winning]) if winning else 0
|
| avg_loss = abs(np.mean([s['profit_pct'] for s in losing])) if losing else 0
|
| pl_ratio = avg_win / avg_loss if avg_loss > 0 else 0
|
|
|
|
|
| if win_rate >= 65 and pl_ratio >= 2.0:
|
| conclusion = f'量化模型在{len(signals_found)}次信号中胜率{win_rate:.1f}%，盈亏比{pl_ratio:.2f}，模型对该股票有效'
|
| elif win_rate >= 55:
|
| conclusion = f'胜率{win_rate:.1f}%尚可，但盈亏比{pl_ratio:.2f}偏低，建议配合其他指标使用'
|
| elif win_rate >= 45:
|
| conclusion = f'胜率{win_rate:.1f}%，盈亏比{pl_ratio:.2f}，模型对该股票区分度一般'
|
| else:
|
| conclusion = f'胜率{win_rate:.1f}%，盈亏比{pl_ratio:.2f}，量化模型对该股票效果不佳，谨慎使用'
|
|
|
| return {
|
| 'total_signals': len(signals_found),
|
| 'win_count': len(winning),
|
| 'lose_count': len(losing),
|
| 'win_rate': round(win_rate, 1),
|
| 'avg_profit_pct': round(avg_win, 2),
|
| 'avg_loss_pct': round(avg_loss, 2),
|
| 'profit_loss_ratio': round(pl_ratio, 2),
|
| 'hold_days': hold_days,
|
| 'sample_signals': signals_found[-10:],
|
| 'conclusion': conclusion,
|
| }
|
|
|
