#!/usr/bin/env python3 """ 分析ETH过去4小时的市场数据，检查为什么没有开仓 """ import sys import os import asyncio from datetime import datetime, timedelta # 添加项目路径 sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'backend')) from app.services.binance_service import binance_service from app.crypto_agent.market_signal_analyzer import MarketSignalAnalyzer from app.crypto_agent.trading_decision_maker import TradingDecisionMaker from app.services.paper_trading_service import get_paper_trading_service from app.utils.logger import logger async def analyze_eth_4h(): """分析ETH过去4小时的情况""" print("=" * 80) print("📊 ETH 过去4小时市场分析") print("=" * 80) print(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") # 1. 获取服务 binance = binance_service signal_analyzer = MarketSignalAnalyzer() decision_maker = TradingDecisionMaker() trading_service = get_paper_trading_service() # 2. 获取当前行情 print("\n📈 当前行情:") current_price = binance.get_current_price('ETHUSDT') stats = binance.get_24h_stats('ETHUSDT') if current_price: print(f" 价格: ${current_price:,.2f}") if stats: print(f" 涨跌: {stats.get('priceChangePercent', 0):+.2f}%") print(f" 成交量: {stats.get('volume', 0):,.0f}") # 3. 获取K线数据（多周期） print("\n📊 获取K线数据...") data = binance.get_multi_timeframe_data('ETHUSDT') if not data: print("❌ 无法获取K线数据") return # 显示各周期数据量 for tf, df in data.items(): print(f" {tf}: {len(df)} 条数据") # 4. 检查过去4小时的价格波动 print("\n📊 过去4小时价格波动:") df_15m = data.get('15m') if df_15m is not None and len(df_15m) > 0: # 获取过去4小时的15分钟数据（16根） recent_4h = df_15m.tail(16) if len(recent_4h) > 0: high = recent_4h['high'].max() low = recent_4h['low'].min() start_price = recent_4h.iloc[0]['close'] end_price = recent_4h.iloc[-1]['close'] volatility = ((high - low) / start_price) * 100 price_change = ((end_price - start_price) / start_price) * 100 print(f" 最高: ${high:,.2f}") print(f" 最低: ${low:,.2f}") print(f" 波动幅度: {volatility:.2f}%") print(f" 价格变化: {price_change:+.2f}%") # 5. 运行信号分析 print("\n🔍 信号分析:") print("-" * 80) analysis = await signal_analyzer.analyze('ETHUSDT', data) # 显示分析结果 print(f"\n市场状态: {analysis.get('analysis_summary', 'N/A')}") # 新闻情绪 if analysis.get('news_sentiment'): sentiment_map = {'positive': '📈 积极', 'negative': '📉 消极', 'neutral': '➖ 中性'} print(f"新闻情绪: {sentiment_map.get(analysis['news_sentiment'], analysis['news_sentiment'])}") # 信号 signals = analysis.get('signals', []) if signals: print(f"\n🎯 生成 {len(signals)} 个信号:") for sig in signals: action = sig.get('action', 'wait') action_map = {'buy': '🟢 做多', 'sell': '🔴 做空', 'wait': '⏸️ 观望'} grade = sig.get('grade', 'D') confidence = sig.get('confidence', 0) print(f"\n {action_map.get(action, action)} [{grade}级] {confidence}%") print(f" 入场: ${sig.get('entry_price', 0):,.2f}") print(f" 止损: ${sig.get('stop_loss', 0):,.2f}") print(f" 止盈: ${sig.get('take_profit', 0):,.2f}") print(f" 入场方式: {sig.get('entry_type', 'N/A')}") if sig.get('reason'): print(f" 理由: {sig['reason']}") if sig.get('risk_warning'): print(f" ⚠️ 风险: {sig['risk_warning']}") else: print("\n⏸️ 无交易信号") # 6. 检查当前持仓 print("\n💼 当前持仓:") account = trading_service.get_account_status() active_orders = trading_service.get_active_orders() print(f" 余额: ${account['current_balance']:,.2f}") print(f" 已用保证金: ${account['used_margin']:,.2f}") print(f" 持仓数量: {len(active_orders)}") if active_orders: for order in active_orders: print(f"\n {order.symbol} {order.side.value}") print(f" 入场价: ${order.filled_price:,.2f}") print(f" 当前价: ${current_price:,.2f}") print(f" 保证金: ${order.margin:,.2f}") print(f" 杠杆: {order.leverage}x") else: print(" 无持仓") # 7. 模拟交易决策 print("\n🤖 交易决策分析:") print("-" * 80) if signals: # 对每个信号做决策 for sig in signals[:3]: # 只看前3个 decision = await decision_maker.make_decision( symbol='ETHUSDT', signals=[sig], current_data=data, current_price=current_price ) print(f"\n信号: {sig.get('action', 'wait')} | {sig.get('grade')}级 | {sig.get('confidence')}%") print(f"决策: {decision['decision']}") print(f"理由: {decision.get('reason', 'N/A')}") if decision.get('orders_to_open'): print(f"开仓: {len(decision['orders_to_open'])} 个") if decision.get('orders_to_close'): print(f"平仓: {decision['orders_to_close']}") else: print("无信号，不生成决策") # 8. 检查是否有追涨杀跌的情况 print("\n⚠️ 追涨杀跌检测:") print("-" * 80) if df_15m is not None and len(df_15m) >= 5: recent = df_15m.tail(5) # 计算RSI import pandas_ta as ta rsi = recent['close'].ta.rsi(length=14) current_rsi = rsi.iloc[-1] if not rsi.empty else 50 # 检查连续大阳线/阴线 big_moves = 0 for i in range(1, min(4, len(recent))): change = abs((recent.iloc[i]['close'] - recent.iloc[i-1]['close']) / recent.iloc[i-1]['close']) * 100 if change >= 1.0: big_moves += 1 # 检查价格偏离EMA ema5 = recent['close'].ta.ema(length=5) current_price = recent.iloc[-1]['close'] deviation = abs((current_price - ema5.iloc[-1]) / ema5.iloc[-1]) * 100 if not ema5.empty else 0 print(f" RSI(14): {current_rsi:.1f}") print(f" 连续大K线: {big_moves} 根") print(f" 偏离EMA5: {deviation:.2f}%") # 判断是否满足追涨杀跌条件 is_chasing = False reasons = [] if current_rsi > 65 or current_rsi < 35: is_chasing = True reasons.append(f"RSI处于极端区间 ({current_rsi:.1f})") if big_moves >= 2: is_chasing = True reasons.append(f"连续{big_moves}根大K线") if deviation > 1.5: is_chasing = True reasons.append(f"价格偏离EMA5超过1.5%") if is_chasing: print(f"\n 🚨 检测到追涨杀跌条件！") for reason in reasons: print(f" - {reason}") print(f"\n → 这是防止开仓的原因之一") else: print(f"\n ✅ 未检测到追涨杀跌条件") print("\n" + "=" * 80) print("分析完成") print("=" * 80) if __name__ == "__main__": try: asyncio.run(analyze_eth_4h()) except Exception as e: print(f"\n❌ 分析失败: {e}") import traceback traceback.print_exc()