""" Комбинированный анализ: MoE (ML) → фундаментальный контекст → вердикт. Flow: 0. Проверка последних свечей на артефакты (спайки) 1. MoE предсказание для всех 14 тикеров MOEX 2. Для тикеров с BUY/SELL сигналом — сегментный анализ 3. Финальный вердикт: подтверждает или отклоняет сигнал """ from typing import Any, Optional import os import numpy as np import pandas as pd from config import MOEX_TICKERS from models.moe import MultiTimeframeMoE from analysis.segments import ( get_segment_for_ticker, get_segment_name_ru, get_segment_context_for_ticker, MOEX_SEGMENTS, ) SAVE_DIR = 'models/saved' MODEL_VERSIONS = ['_moe_v125.joblib', '_moe_v12.joblib', '_moe_v11.joblib'] # search order: newest first def _find_model_path(ticker: str) -> str | None: """Ищет последовательно v12, затем v11.""" for suffix in MODEL_VERSIONS: p = f'{SAVE_DIR}/{ticker.lower()}{suffix}' if os.path.exists(p): return p return None def run_moe_predictions() -> dict[str, dict[str, Any]]: """ MoE предсказание для всех 14 тикеров. Returns: {ticker: {signal, confidence, p_long, p_short, val_acc}} """ import os results = {} for ticker in MOEX_TICKERS: path = _find_model_path(ticker) if path is None: results[ticker] = {'signal': 'NO_MODEL', 'confidence': 0.0} continue try: moe = MultiTimeframeMoE(ticker) moe.load(path) pred = moe.predict(limit=1000) results[ticker] = { 'signal': pred['signal'], 'confidence': pred['confidence'], 'p_long': pred['p_long'], 'p_short': pred['p_short'], 'val_acc': moe.val_acc, 'model_version': 'v12' if '_v12' in path else 'v11', } except Exception as e: results[ticker] = {'signal': 'ERROR', 'confidence': 0.0, 'error': str(e)} return results def get_fundamental_context(ticker: str) -> dict[str, Any]: """ Фундаментальный контекст для тикера: - сегмент и его анализ по W1/D1/H1 """ ctx = get_segment_context_for_ticker(ticker) return ctx def generate_verdict( moe_result: dict[str, Any], fund_ctx: dict[str, Any], ) -> dict[str, Any]: """ Генерация финального вердикта на основе MoE + фундаментального контекста. Правила: - Если сегментный тренд совпадает с сигналом MoE → CONFIRMED (уверенность растёт) - Если сегментный тренд противоположен → CONFLICT (сигнал ослабляется) - Если сегментного контекста нет → MoE standalone - Если MoE сигнал NEUTRAL → пропускаем """ signal = moe_result.get('signal', 'NEUTRAL') confidence = moe_result.get('confidence', 0.0) p_long = moe_result.get('p_long', 0.5) p_short = moe_result.get('p_short', 0.5) if signal == 'NEUTRAL' or signal in ('NO_MODEL', 'ERROR'): return { 'verdict': signal, 'confidence': confidence, 'explanation': 'MoE сигнала нет', } segment = fund_ctx.get('segment') if segment is None: return { 'verdict': signal, 'confidence': confidence, 'explanation': 'Тикер не привязан к сегменту', } summary = fund_ctx.get('summary', {}) composite_trend = summary.get('composite_trend', 'sideways') trend_alignment = summary.get('trend_alignment', 'unknown') avg_rsi = summary.get('avg_rsi') analysis = fund_ctx.get('analysis', {}) name_ru = fund_ctx.get('name_ru', segment) # Определяем желаемое направление сигнала signal_direction = 'up' if signal == 'BUY' else 'down' # Собираем факты facts = [] verdict = signal adj_confidence = confidence # 1. Совпадение тренда if composite_trend == signal_direction: facts.append(f'сегмент "{name_ru}" в том же направлении ({composite_trend})') adj_confidence = min(confidence * 1.3, 0.95) elif composite_trend == 'sideways': facts.append(f'сегмент "{name_ru}" без ярко выраженного тренда') adj_confidence = confidence # не меняем else: facts.append(f'сегмент "{name_ru}" в противоположном направлении ({composite_trend})') adj_confidence = confidence * 0.6 verdict = 'CONFLICT' # 2. RSI контекст if avg_rsi is not None: if signal == 'BUY' and avg_rsi < 40: facts.append(f'RSI сегмента {avg_rsi} — перепроданность, лонг подтверждается') adj_confidence = min(adj_confidence * 1.15, 0.95) elif signal == 'SELL' and avg_rsi > 60: facts.append(f'RSI сегмента {avg_rsi} — перекупленность, шорт подтверждается') adj_confidence = min(adj_confidence * 1.15, 0.95) elif signal == 'BUY' and avg_rsi > 70: facts.append(f'RSI сегмента {avg_rsi} — перекупленность, сигнал BUY под вопросом') adj_confidence *= 0.8 elif signal == 'SELL' and avg_rsi < 30: facts.append(f'RSI сегмента {avg_rsi} — перепроданность, сигнал SELL под вопросом') adj_confidence *= 0.8 # 3. Согласованность таймфреймов n_tfs = summary.get('n_tfs_with_data', 0) if n_tfs >= 2 and composite_trend == signal_direction: aligned_tfs = sum( 1 for a in analysis.values() if a.get('has_data') and a.get('trend') == signal_direction ) if aligned_tfs >= 2: facts.append(f'тренд подтверждён на {aligned_tfs}/{n_tfs} таймфреймах') adj_confidence = min(adj_confidence * 1.1, 0.95) return { 'verdict': verdict, 'confidence': round(float(adj_confidence), 3), 'original_signal': signal, 'original_confidence': round(float(confidence), 3), 'segment': segment, 'segment_name_ru': name_ru, 'composite_trend': composite_trend, 'segment_rsi': avg_rsi, 'facts': facts, 'explanation': '; '.join(facts) if facts else 'нет дополнительных факторов', } def check_for_artifacts(): """ Проверка последних свечей всех тикеров на артефакты/спайки перед анализом. Если обнаружены аномалии — автоматическая очистка через data.cleaner. Работает только для тикеров, где известны ценовые пороги (SNGSP, PLZL). """ from data.loader import load_dataframe from data.cleaner import clean_ticker_in_db, PRICE_THRESHOLDS print(f'\n [0/2] Проверка данных на артефакты...') cleaned_any = False for ticker in MOEX_TICKERS: # Быстрая проверка последних свечей try: df = load_dataframe(ticker, 'H1', limit=5) for col in ['Open', 'High', 'Low', 'Close']: df[col] = pd.to_numeric(df[col], errors='coerce') price = float(df['Close'].iloc[-1]) # Проверка по известным ценовым порогам if ticker in PRICE_THRESHOLDS: min_price = PRICE_THRESHOLDS[ticker]['min_price'] if price < min_price: print(f' ⚠️ {ticker}: цена {price:.2f} < {min_price} — артефакт! Очистка...') n = clean_ticker_in_db(ticker, 'H1', dry_run=False) print(f' ✅ {ticker}: удалено {n} спайков') cleaned_any = True continue # Универсальная проверка: свеча отличается от предыдущей >500% if len(df) >= 2: prev = float(df['Close'].iloc[-2]) if prev > 0 and price > 0: ratio = price / prev if price > prev else prev / price if ratio > 5.0: print(f' ⚠️ {ticker}: цена {prev:.2f} → {price:.2f} (×{ratio:.1f}) — возможен артефакт') n = clean_ticker_in_db(ticker, 'H1', dry_run=False) print(f' ✅ {ticker}: удалено {n} спайков') cleaned_any = True except Exception as e: print(f' ⚠️ {ticker}: ошибка проверки — {e}') if not cleaned_any: print(f' ✅ Все тикеры чисты') def analyze_all() -> list[dict[str, Any]]: """ Полный анализ всех тикеров: MoE → фундаментальный контекст → вердикт. """ print(f'\n{"=" * 70}') print(f' КОМБИНИРОВАННЫЙ АНАЛИЗ: MoE + Фундаментальный контекст') print(f'{"=" * 70}') # Шаг 0: Проверка артефактов check_for_artifacts() # Шаг 1: MoE предсказания print(f'\n [1/2] MoE предсказания...') moe_results = run_moe_predictions() n_signal = sum(1 for r in moe_results.values() if r.get('signal') in ('BUY', 'SELL')) print(f' Найдено {n_signal} тикеров с сигналом из {len(MOEX_TICKERS)}') # Шаг 2: Фундаментальный контекст + вердикт print(f'\n [2/2] Фундаментальный анализ...') all_results = [] for ticker in MOEX_TICKERS: moe_r = moe_results.get(ticker, {}) signal = moe_r.get('signal', 'NEUTRAL') # Фундаментальный контекст для всех тикеров (даже нейтральных) fund_ctx = get_fundamental_context(ticker) # Вердикт только для сигнальных if signal in ('BUY', 'SELL'): verdict = generate_verdict(moe_r, fund_ctx) else: verdict = { 'verdict': signal, 'original_signal': signal, 'original_confidence': moe_r.get('confidence', 0.0), 'segment': fund_ctx.get('segment'), 'segment_name_ru': fund_ctx.get('name_ru'), } verdict['ticker'] = ticker verdict['segment_name_ru'] = fund_ctx.get('name_ru', '—') verdict['val_acc'] = moe_r.get('val_acc', 0.0) all_results.append(verdict) return all_results def print_analysis(results: list[dict[str, Any]]): """Форматированный вывод результатов анализа.""" print(f'\n{"=" * 90}') print(f' ИТОГОВЫЙ ВЕРДИКТ') print(f'{"=" * 90}') # Сигнальные тикеры print(f'\n 🟢🔴 СИГНАЛЫ:') for r in results: if r.get('verdict') in ('BUY', 'SELL', 'CONFLICT'): ticker = r['ticker'] verdict = r['verdict'] conf = r.get('confidence', 0) orig_conf = r.get('original_confidence', 0) val_acc = r.get('val_acc', 0) seg = r.get('segment_name_ru', '—') trend = r.get('composite_trend', '—') rsi = r.get('segment_rsi', '—') facts = r.get('facts', []) icon = {'BUY': '🟢', 'SELL': '🔴', 'CONFLICT': '⚠️'}.get(verdict, '⚪') orig = r.get('original_signal', verdict) print(f'\n {icon} {ticker} ({seg})') print(f' MoE: {orig:5s} (conf={orig_conf:.1%}) | val_acc={val_acc:.2%}') print(f' Сегмент: trend={trend}, RSI={rsi}' if isinstance(rsi, str) or rsi is not None else '') print(f' Вердикт: {verdict} (conf={conf:.1%})') if facts: for f in facts: print(f' • {f}') # Нейтральные тикеры neutrals = [r for r in results if r.get('verdict') == 'NEUTRAL'] no_model = [r for r in results if r.get('verdict') == 'NO_MODEL'] errors = [r for r in results if r.get('verdict') == 'ERROR'] if neutrals: print(f'\n ⚪ НЕЙТРАЛЬНЫЕ ({len(neutrals)}):') print(f' {", ".join(r["ticker"] for r in neutrals)}') if no_model: print(f'\n ❌ НЕТ МОДЕЛИ ({len(no_model)}):') print(f' {", ".join(r["ticker"] for r in no_model)}') if errors: print(f'\n ❌ ОШИБКИ ({len(errors)}):') for r in errors: print(f' {r["ticker"]}: {r.get("explanation", "?")}') def add_trade_plans( results: list[dict[str, Any]], capital: float = 100_000.0, risk_percent: float = 1.0, min_confidence: float = 0.70, ) -> list[dict[str, Any]]: """ Для CONFIRMED сигналов (BUY/SELL с высокой уверенностью) — расчёт параметров сделки через models.trade_planner.plan_trade. Args: results: список вердиктов из analyze_all() capital: бюджет в рублях risk_percent: риск на сделку, % от капитала min_confidence: мин. уверенность для расчёта сделки Returns: тот же список с добавленным полем 'trade_plan' у сигнальных тикеров """ from models.trade_planner import plan_trade for r in results: verdict = r.get('verdict', '') confidence = r.get('confidence', 0.0) ticker = r.get('ticker', '') # Только CONFIRMED (BUY/SELL, не CONFLICT, с высокой уверенностью) if verdict in ('BUY', 'SELL') and confidence >= min_confidence: try: trade_plan = plan_trade( ticker=ticker, direction=verdict, confidence=confidence, capital=capital, risk_percent=risk_percent, ) r['trade_plan'] = trade_plan if 'error' in trade_plan: r['trade_plan_error'] = trade_plan['error'] except Exception as e: r['trade_plan_error'] = str(e) return results def print_trade_plans(results: list[dict[str, Any]]): """Вывод trade_plan для сигнальных тикеров.""" plans = [r for r in results if r.get('trade_plan') and 'error' not in r.get('trade_plan', {})] if not plans: return print(f'\n{"=" * 90}') print(f' 📋 ПЛАНЫ СДЕЛОК (CONFIRMED)') print(f'{"=" * 90}') for r in plans: tp = r['trade_plan'] ticker = r['ticker'] direction = tp['direction'] conf = tp['confidence'] icon = '🟢' if direction == 'BUY' else '🔴' print(f'\n {icon} {ticker} | {direction} (conf={conf:.1%}) | val_acc={r.get("val_acc", 0):.2%}') print(f' {"─" * 80}') print(f' Entry: {tp["entry"]:>10.2f} (текущая цена: {tp["current_price"]:.2f})') print(f' Stop Loss: {tp["stop_loss"]:>10.2f} ({tp["sl_pct"]:.2f}%)') print(f' Take Profit:{tp["take_profit"]:>10.2f} ({tp["tp_pct"]:.2f}%)') print(f' ATR: {tp["atr"]:>10.4f} ({tp["atr_pct"]:.2f}% от цены)') print(f' RR: {tp["rr_ratio"]:>10.2f}') pos = tp['position'] print(f'\n Позиция:') print(f' Капитал: {pos["capital"]:>10.2f} ₽') print(f' Риск: {pos["max_risk_rub"]:>10.2f} ₽ ({pos["risk_percent"]}%)') print(f' Лотов: {pos["lots"]:>10d} × {pos["lot_size"]} шт = {pos["shares"]} шт') print(f' Стоимость: {pos["position_cost"]:>10.2f} ₽ ({pos["position_cost"]/pos["capital"]*100:.1f}% бюджета)') print(f' Факт. риск: {pos["actual_risk"]:>10.2f} ₽ ({pos["risk_pct_of_capital"]:.1f}%)') print(f'\n Лесенка (3 уровня):') print(f' {"Ур.":>4} {"Цена":>10} {"% поз.":>6} {"Нараст.":>8} {"RR":>6}') print(f' {"────":>4} {"──────────":>10} {"──────":>6} {"────────":>8} {"────":>6}') for step in tp['ladder']: print(f' {step["level"]:>4d} {step["price"]:>10.2f} {step["size_pct"]:>5d}% {step["cumulative_pct"]:>7d}% {step["rr"]:>6.2f}') print(f' {"─" * 80}') def analyze_all_with_trade_plans( capital: float = 100_000.0, risk_percent: float = 1.0, min_confidence: float = 0.70, ) -> list[dict[str, Any]]: """ Полный анализ + планы сделок для CONFIRMED сигналов. """ results = analyze_all() results = add_trade_plans( results, capital=capital, risk_percent=risk_percent, min_confidence=min_confidence, ) return results if __name__ == '__main__': results = analyze_all_with_trade_plans() print_analysis(results) print_trade_plans(results)