""" Спекулятивный технический анализ VTBR с сегментным контекстом. """ import sys, os, json sys.path.insert(0, os.path.dirname(__file__)) import pandas as pd import numpy as np from src.db.connection import fetch_ohlcv_combined from src.indicators.calculations import calc_all_indicators from src.analysis.tech_analysis import ( detect_candlestick_patterns, detect_vsa_signals, find_key_levels, determine_trend, wyckoff_phase, generate_trade_signal, ) from src.analysis.segments import ( get_segment_for_ticker, get_segment_context_for_ticker, MOEX_SEGMENTS, ) from src.api.tbank import get_order_book, get_last_price, analyze_order_book, get_orderbook_with_history, get_current_candle TICKER = 'VTBR' FIGI = 'BBG004730ZJ9' def get_last_val(df, col): if col in df.columns and len(df) > 0: v = df[col].iloc[-1] return float(v) if pd.notna(v) else None return None def main(): print(f"=== {TICKER} Спекулятивный анализ ===") # 1. Загружаем данные print("Загрузка данных H1, D1, W1...") df_h1 = fetch_ohlcv_combined(TICKER, 'H1', limit=200) df_d1 = fetch_ohlcv_combined(TICKER, 'D1', limit=200) df_w1 = fetch_ohlcv_combined(TICKER, 'W1', limit=100) print(f"H1: {len(df_h1)} свечей, D1: {len(df_d1)}, W1: {len(df_w1)}") # Текущие свечи for tf_name in ['W1', 'D1', 'H1']: try: cur_candle = get_current_candle(TICKER, tf_name) if cur_candle: print(f" Текущая {tf_name}: {cur_candle['Date']} " f"O={cur_candle['Open']:.2f} H={cur_candle['High']:.2f} " f"L={cur_candle['Low']:.2f} C={cur_candle['Close']:.2f} " f"Vol={cur_candle['Volume']}") except Exception: pass # 2. Индикаторы df_h1 = calc_all_indicators(df_h1) df_d1 = calc_all_indicators(df_d1) df_w1 = calc_all_indicators(df_w1) # 3. Свечные паттерны df_h1 = detect_candlestick_patterns(df_h1) df_d1 = detect_candlestick_patterns(df_d1) # 4. VSA df_h1 = detect_vsa_signals(df_h1) df_d1 = detect_vsa_signals(df_d1) # 5. Уровни levels_h1 = find_key_levels(df_h1) levels_d1 = find_key_levels(df_d1) levels_w1 = find_key_levels(df_w1) # 6. Тренды trend_w1, adx_w1 = determine_trend(df_w1) trend_d1, adx_d1 = determine_trend(df_d1) trend_h1, adx_h1 = determine_trend(df_h1) # 7. Вайкофф wyckoff = wyckoff_phase(df_w1) # 8. Текущая цена try: current_price = get_last_price(TICKER) if current_price: print(f"T-Bank API last price: {current_price}") else: current_price = float(df_h1['Close'].iloc[-1]) except Exception as e: current_price = float(df_h1['Close'].iloc[-1]) print(f"T-Bank API error, fallback to H1 close: {current_price}") # 9. Стакан (live API + файловая история OrderBookMonitor) ob_data = None try: ob_data = get_orderbook_with_history(TICKER, depth=20, max_age_sec=120) live = ob_data['live'] snap = live['snapshot'] file_used = ob_data['file_used'] combined_verdict = ob_data['combined_verdict'] print(f"OrderBook: bid={snap['best_bid']} ask={snap['best_ask']} " f"spread={snap['spread_pct']:.3f}% imbalance={snap['imbalance_ratio']:.2f}") print(f" Файл истории прочитан: {file_used}") print(f" Комбинированный вердикт: {combined_verdict}") if ob_data.get('history') and ob_data['history'].get('bias'): hb = ob_data['history']['bias'] print(f" Bias файла: {hb['label']} (🟢{hb['bullish']} ⚪️{hb['neutral']} 🔴{hb['bearish']})") except Exception as e: print(f"OrderBook error: {e}") # ── 10. СЕГМЕНТНЫЙ КОНТЕКСТ ── seg_name = get_segment_for_ticker(TICKER) segment_ctx = None if seg_name: seg_cfg = MOEX_SEGMENTS.get(seg_name, {}) print(f"\n 📊 Сегмент: {seg_name} ({seg_cfg.get('name_ru', seg_name)})") segment_ctx = get_segment_context_for_ticker(TICKER, tfs=['W1', 'D1', 'H1']) summary = segment_ctx.get('summary', {}) print(f" Композитный тренд: {summary.get('composite_trend', '—')}") print(f" Средний RSI: {summary.get('avg_rsi', '—')}") for tf_name in ['W1', 'D1', 'H1']: tf = segment_ctx.get('analysis', {}).get(tf_name, {}) if tf.get('has_data'): print(f" {tf_name}: trend={tf['trend']} ADX={tf.get('adx','?')} RSI={tf.get('rsi','?')}") # ── 11. Уровни + сигнал ── all_levels = { 'support': list(set(levels_h1.get('support', []) + levels_d1.get('support', []) + levels_w1.get('support', []))), 'resistance': list(set(levels_h1.get('resistance', []) + levels_d1.get('resistance', []) + levels_w1.get('resistance', []))), } all_levels['support'] = sorted([x for x in all_levels['support'] if x < current_price], reverse=True)[:5] all_levels['resistance'] = sorted([x for x in all_levels['resistance'] if x > current_price])[:5] # Build orderbook_bias dict from already-fetched ob_data ob_bias_for_signal = None if ob_data: ob_bias_for_signal = { 'combined_verdict': ob_data.get('combined_verdict', ''), 'live_verdict': ob_data.get('live', {}).get('verdict', ''), 'combined_score': ob_data.get('combined_bias', {}).get('combined_score', 0.0), 'file_bias': ob_data.get('history', {}).get('bias') if ob_data.get('history') else None, 'file_used': ob_data.get('file_used', False), } signal = generate_trade_signal( trend_w1=trend_w1, trend_d1=trend_d1, trend_h1=trend_h1, phase=wyckoff['phase'], levels=all_levels, df_d1=df_d1, df_h1=df_h1, current_price=current_price, segment_context=segment_ctx, orderbook_bias=ob_bias_for_signal, ) # ── ВЫВОД JSON ── result = { 'ticker': TICKER, 'last_price': current_price, 'segment': seg_name, 'segment_context': segment_ctx, 'trend_w1': trend_w1, 'trend_d1': trend_d1, 'trend_h1': trend_h1, 'wyckoff_phase': wyckoff['phase'], 'wyckoff_events': wyckoff['events'], 'wyckoff_tr_low': wyckoff['tr_low'], 'wyckoff_tr_high': wyckoff['tr_high'], 'wyckoff_direction': wyckoff['direction'], 'signal': signal['signal'], 'confidence': signal['confidence'], 'entry': signal['entry'], 'sl': signal['sl'], 'tp': signal['tp'], 'reason': signal['reason'], 'atr': signal['atr'], 'levels_support': all_levels['support'], 'levels_resistance': all_levels['resistance'], } if ob_data: result['orderbook'] = { **ob_data['live']['snapshot'], 'bid_walls': ob_data['live']['bid_walls'], 'ask_walls': ob_data['live']['ask_walls'], 'support_zones': ob_data['live']['support_zones'], 'resistance_zones': ob_data['live']['resistance_zones'], 'verdict': ob_data['live']['verdict'], 'live_verdict': ob_data['live']['verdict'], 'details': ob_data['live']['details'], 'file_used': ob_data['file_used'], 'combined_verdict': ob_data['combined_verdict'], 'combined_score': ob_data.get('combined_bias', {}).get('combined_score', 0.0), 'file_bias': ob_data['history']['bias'] if ob_data.get('history') and ob_data['history'].get('bias') else None, 'file_interpretations_count': len(ob_data['history'].get('interpretations', [])) if ob_data.get('history') else 0, } result['indicators'] = { 'h1': { 'close': get_last_val(df_h1, 'Close'), 'ema_9': get_last_val(df_h1, 'EMA_9'), 'ema_21': get_last_val(df_h1, 'EMA_21'), 'ema_50': get_last_val(df_h1, 'EMA_50'), 'rsi': get_last_val(df_h1, 'RSI_14'), 'macd_hist': get_last_val(df_h1, 'MACD_hist'), 'adx': get_last_val(df_h1, 'ADX_14'), 'plus_di': get_last_val(df_h1, 'Plus_DI'), 'minus_di': get_last_val(df_h1, 'Minus_DI'), 'atr': get_last_val(df_h1, 'ATR_14'), 'volume': get_last_val(df_h1, 'Volume'), }, 'd1': { 'close': get_last_val(df_d1, 'Close'), 'ema_9': get_last_val(df_d1, 'EMA_9'), 'ema_21': get_last_val(df_d1, 'EMA_21'), 'ema_50': get_last_val(df_d1, 'EMA_50'), 'rsi': get_last_val(df_d1, 'RSI_14'), 'macd_hist': get_last_val(df_d1, 'MACD_hist'), 'adx': get_last_val(df_d1, 'ADX_14'), 'plus_di': get_last_val(df_d1, 'Plus_DI'), 'minus_di': get_last_val(df_d1, 'Minus_DI'), 'atr': get_last_val(df_d1, 'ATR_14'), 'volume': get_last_val(df_d1, 'Volume'), }, 'w1': { 'close': get_last_val(df_w1, 'Close'), 'ema_50': get_last_val(df_w1, 'EMA_50'), 'ema_200': get_last_val(df_w1, 'EMA_200'), 'rsi': get_last_val(df_w1, 'RSI_14'), 'macd_hist': get_last_val(df_w1, 'MACD_hist'), 'adx': get_last_val(df_w1, 'ADX_14'), 'plus_di': get_last_val(df_w1, 'Plus_DI'), 'minus_di': get_last_val(df_w1, 'Minus_DI'), }, } # Последние 5 свечей H1 h1_last_5 = [] for i in range(max(0, len(df_h1)-5), len(df_h1)): row = df_h1.iloc[i] h1_last_5.append({ 'date': str(row.get('Date', '')), 'time': str(row.get('Time', '')), 'open': float(row['Open']), 'high': float(row['High']), 'low': float(row['Low']), 'close': float(row['Close']), 'volume': int(row.get('Volume', 0)), 'vsa': str(row.get('VSA_signal', '')), 'candle': str(row.get('candle_pattern', '')), }) result['h1_last_5_candles'] = h1_last_5 d1_last_3 = [] for i in range(max(0, len(df_d1)-3), len(df_d1)): row = df_d1.iloc[i] d1_last_3.append({ 'date': str(row.get('Date', '')), 'open': float(row['Open']), 'high': float(row['High']), 'low': float(row['Low']), 'close': float(row['Close']), 'volume': int(row.get('Volume', 0)), 'vsa': str(row.get('VSA_signal', '')), 'candle': str(row.get('candle_pattern', '')), }) result['d1_last_3_candles'] = d1_last_3 print("\n=== JSON RESULT ===") print(json.dumps(result, indent=2, ensure_ascii=False, default=str)) if __name__ == '__main__': main()