""" Историческое тестирование Wyckoff D1 модели (walk-forward) на одном тикере. Алгоритм: 1. Загружаем ВСЕ данные D1 для тикера 2. В каждой тестовой точке (каждые N баров): - Обрезаем данные до этой даты (нет lookahead) - Прогоняем через NN модель (feature prep + inference) - Записываем фазу + confidence 3. Смотрим forward return (цена через M баров) 4. Сравниваем: фаза предсказала движение цены? Запуск: python _backtest_wyckoff_history.py [--ticker X5] [--step 10] [--forward 20] """ import sys import json import warnings import argparse from pathlib import Path from collections import Counter warnings.filterwarnings("ignore") sys.path.insert(0, str(Path(__file__).resolve().parent)) import pandas as pd import numpy as np from src.db.connection import fetch_ohlcv_combined from src.ml.inference.wyckoff_inference import WyckoffInference # ── Map phase IDs → readable names ── PHASE_NAMES = { 0: 'Маркдаун', 1: 'Накопление (раннее)', 2: 'Накопление (позд)/Маркап', 3: 'Маркап', 4: 'Распределение', } PHASE_GROUP = { 0: '🔴 Медвежий', 1: '🟢 Бычий', 2: '🟢 Бычий', 3: '🟢 Бычий', 4: '🔴 Медвежий', } # Предсказание направления: Accumulation(1,2) и Markup(3) → бычьи, Markdown(0) и Distribution(4) → медвежьи PHASE_DIRECTION = {0: -1, 1: 1, 2: 1, 3: 1, 4: -1} def run_backtest(ticker: str, step: int = 10, forward_bars: int = 20, min_bars: int = 300): """ Walk-forward backtest Wyckoff D1 модели. Args: ticker: тикер MOEX. step: шаг между тестовыми точками (в свечах D1). forward_bars: горизонт forward return (в свечах D1). min_bars: минимальное количество баров для начала теста. """ print(f"{'=' * 90}") print(f" ИСТОРИЧЕСКОЕ ТЕСТИРОВАНИЕ WYCKOFF — {ticker}") print(f" Шаг: {step} дн | Forward: {forward_bars} дн | Мин.баров: {min_bars}") print(f"{'=' * 90}") # ── 1. Загружаем все D1 данные ── print(f"\n[1/4] Загрузка {ticker} D1 из БД...") df_all = fetch_ohlcv_combined(ticker, 'D1', limit=5000) if df_all is None or len(df_all) < min_bars: print(f" ❌ Недостаточно данных: {len(df_all) if df_all is not None else 0} < {min_bars}") return # Сортируем по возрастанию (исторический порядок) df_all = df_all.sort_values('timestamp').reset_index(drop=True) print(f" ✅ Загружено {len(df_all)} свечей ({df_all['Date'].iloc[0]} — {df_all['Date'].iloc[-1]})") # ── 2. Загружаем модель один раз ── print(f"\n[2/4] Загрузка D1 модели (wyckoff_mt_d1_lstm_v2)...") inferer = WyckoffInference( model_name='wyckoff_mt_d1_lstm_v2', ticker=ticker, tf='D1', use_rule_based_fallback=True, ) if not inferer.model_loaded: print(" ❌ Модель не загружена") return print(f" ✅ Модель загружена (seq_len={inferer.metadata.get('seq_len', '?')})") # ── 3. Walk-forward: идём по истории ── print(f"\n[3/4] Walk-forward тестирование (каждые {step} баров, forward {forward_bars})...") test_points = list(range(min_bars, len(df_all), step)) results = [] errors = 0 for idx, test_idx in enumerate(test_points): # Текущая дата и цена current_bar = df_all.iloc[test_idx] current_date = current_bar['Date'] current_close = float(current_bar['Close']) # Обрезаем данные ДО этой точки (строго исторически) df_truncated = df_all.iloc[:test_idx + 1].copy() try: # Вызываем модель result = inferer.analyze(ticker=ticker, df=df_truncated) # Forward return: цена через forward_bars future_idx = min(test_idx + forward_bars, len(df_all) - 1) future_close = float(df_all.iloc[future_idx]['Close']) forward_return = (future_close / current_close - 1) * 100 # Абсолютное движение abs_return = abs(future_close - current_close) / current_close * 100 # Правильно ли предсказано направление? pred_direction = PHASE_DIRECTION.get(result.phase, 0) actual_direction = 1 if forward_return > 0 else -1 if forward_return < 0 else 0 direction_correct = (pred_direction * actual_direction) > 0 results.append({ 'idx': test_idx, 'date': current_date, 'price': round(current_close, 2), 'phase': int(result.phase), 'phase_name': result.phase_name, 'confidence': round(result.confidence, 1), 'probs': {str(k): round(v, 3) for k, v in result.phase_probs.items()}, 'forward_return': round(forward_return, 2), 'abs_return': round(abs_return, 2), 'direction_correct': direction_correct, 'future_idx': future_idx, }) # Прогресс if (idx + 1) % max(1, len(test_points) // 10) == 0: pct = (idx + 1) / len(test_points) * 100 print(f" {pct:.0f}% | {current_date} | Phase={result.phase_name} | " f"Conf={result.confidence:.0f}% | Fwd={forward_return:+.1f}% | " f"{'✓' if direction_correct else '✗'}") except Exception as e: errors += 1 if errors <= 3: print(f" ⚠️ {current_date}: ошибка — {e}") print(f"\n ✅ Тестирование завершено: {len(results)} точек, {errors} ошибок") # ── 4. Анализ результатов ── print(f"\n[4/4] Анализ результатов...\n") if not results: print(" Нет результатов для анализа.") return df_results = pd.DataFrame(results) # Статистика по фазам phase_counts = Counter(r['phase'] for r in results) print(f"\n{'=' * 90}") print(f" РАСПРЕДЕЛЕНИЕ ФАЗ ЗА ПЕРИОД") print(f"{'=' * 90}") print(f"{'Фаза':<35} {'Кол-во':<10} {'%':<8} {'Ср.Conf':<10} {'Ср.FwdRet':<12} {'WinRate':<10}") print("-" * 85) for phase_id in sorted(phase_counts.keys()): subset = df_results[df_results['phase'] == phase_id] count = len(subset) avg_conf = subset['confidence'].mean() avg_fwd = subset['forward_return'].mean() win_rate = subset['direction_correct'].mean() * 100 pct = count / len(results) * 100 print(f"{PHASE_NAMES.get(phase_id, f'Phase {phase_id}'):<35} " f"{count:<10} {pct:<8.1f} " f"{avg_conf:<10.1f} {avg_fwd:<+12.2f} {win_rate:<10.1f}") # Общая статистика total_win = df_results['direction_correct'].sum() total = len(df_results) avg_fwd_all = df_results['forward_return'].mean() avg_conf_all = df_results['confidence'].mean() print(f"\n{'ИТОГО':<35} {total:<10} {100:<8.1f} " f"{avg_conf_all:<10.1f} {avg_fwd_all:<+12.2f} {total_win/total*100:<10.1f}") # ── Матрица правильности по фазам ── print(f"\n{'=' * 90}") print(f" ПРАВИЛЬНОСТЬ НАПРАВЛЕНИЯ ПО ФАЗАМ") print(f"{'=' * 90}") print(f"{'Фаза':<35} {'Бычьих':<10} {'Медвежьих':<12} {'Правильно':<12} {'Точность':<10}") print("-" * 79) for phase_id in sorted(phase_counts.keys()): subset = df_results[df_results['phase'] == phase_id] correct = subset['direction_correct'].sum() total_p = len(subset) acc = correct / total_p * 100 if total_p > 0 else 0 if PHASE_DIRECTION.get(phase_id, 0) > 0: dir_label = f"↑ {PHASE_NAMES.get(phase_id, '')}" else: dir_label = f"↓ {PHASE_NAMES.get(phase_id, '')}" print(f"{dir_label:<35} " f"{(subset['forward_return'] > 0).sum():<10} " f"{(subset['forward_return'] < 0).sum():<12} " f"{int(correct):<12} {acc:<10.1f}") # ── Анализ по confidence ── print(f"\n{'=' * 90}") print(f" КАЧЕСТВО ПО УРОВНЯМ УВЕРЕННОСТИ") print(f"{'=' * 90}") conf_buckets = [(0, 40), (40, 55), (55, 70), (70, 85), (85, 101)] print(f"{'Уровень':<20} {'Кол-во':<10} {'WinRate':<10} {'Ср.Fwd':<12}") print("-" * 52) for lo, hi in conf_buckets: subset = df_results[(df_results['confidence'] >= lo) & (df_results['confidence'] < hi)] if len(subset) == 0: continue wr = subset['direction_correct'].mean() * 100 avg_fwd = subset['forward_return'].mean() print(f"{lo:.0f}–{hi:.0f}%{'':<15} {len(subset):<10} {wr:<10.1f} {avg_fwd:<+12.2f}") # ── 5. Финальная оценка ── print(f"\n{'=' * 90}") print(f" ВЫВОДЫ") print(f"{'=' * 90}") avg_fwd = df_results['forward_return'].mean() buy_and_hold = (float(df_all['Close'].iloc[-1]) / float(df_all['Close'].iloc[min_bars]) - 1) * 100 print(f"\n 📈 Buy & Hold за период: {buy_and_hold:+.1f}%") print(f" 📊 Средняя доходность прогнозов (forward {forward_bars}д): {avg_fwd:+.2f}%") print(f" 🎯 Общая точность направления: {total_win / total * 100:.1f}% ({int(total_win)}/{total})") # Анализ: если бы торговали по фазам print(f"\n 💡 Если бы торговали только по Accumulation (классы 1,2):") acc_subset = df_results[df_results['phase'].isin([1, 2])] if len(acc_subset) > 0: acc_win = acc_subset['direction_correct'].mean() * 100 acc_avg = acc_subset['forward_return'].mean() print(f" Сигналов: {len(acc_subset)}, WinRate: {acc_win:.1f}%, Средняя доходность: {acc_avg:+.2f}%") print(f"\n 💡 Если бы торговали только по Distribution (класс 4):") dist_subset = df_results[df_results['phase'] == 4] if len(dist_subset) > 0: dist_win = dist_subset['direction_correct'].mean() * 100 dist_avg = dist_subset['forward_return'].mean() print(f" Сигналов: {len(dist_subset)}, WinRate: {dist_win:.1f}%, Средняя доходность: {dist_avg:+.2f}%") # ── Сохраняем результаты ── output = { 'ticker': ticker, 'config': {'step': step, 'forward_bars': forward_bars, 'min_bars': min_bars}, 'results': [ {k: v for k, v in r.items() if k != 'probs'} for r in results ], 'summary': { 'total_points': total, 'avg_confidence': round(avg_conf_all, 1), 'avg_forward_return': round(avg_fwd, 2), 'direction_accuracy': round(total_win / total * 100, 1), 'buy_and_hold_return': round(buy_and_hold, 2), 'accumulation_signals': int(len(acc_subset)) if len(acc_subset) > 0 else 0, 'accumulation_winrate': round(acc_win, 1) if len(acc_subset) > 0 else 0, 'accumulation_avg_return': round(acc_avg, 2) if len(acc_subset) > 0 else 0, 'distribution_signals': int(len(dist_subset)) if len(dist_subset) > 0 else 0, 'distribution_winrate': round(dist_win, 1) if len(dist_subset) > 0 else 0, 'distribution_avg_return': round(dist_avg, 2) if len(dist_subset) > 0 else 0, } } output_path = Path(__file__).resolve().parent / 'reports' / f'wyckoff_backtest_{ticker}.json' with open(output_path, 'w', encoding='utf-8') as f: json.dump(output, f, indent=2, ensure_ascii=False) print(f"\n📁 Результаты сохранены: {output_path}") print() if __name__ == '__main__': parser = argparse.ArgumentParser(description='Backtest Wyckoff D1 модели') parser.add_argument('--ticker', default='X5', help='Тикер MOEX') parser.add_argument('--step', type=int, default=10, help='Шаг между тестами (баров)') parser.add_argument('--forward', type=int, default=20, help='Горизонт forward return (баров)') parser.add_argument('--min-bars', type=int, default=300, help='Минимальное кол-во баров для старта') args = parser.parse_args() run_backtest(args.ticker, args.step, args.forward, args.min_bars)