stock-ai-agent/backend/app/crypto_agent/market_signal_analyzer.py
2026-02-24 01:50:56 +08:00

596 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
市场信号分析器 - 纯市场分析,不包含任何仓位信息
职责:
1. 分析K线、量价、技术指标
2. 分析新闻舆情
3. 输出纯市场信号buy/sell/hold + confidence + reasoning
不负责:
- 仓位管理
- 风险控制
- 具体下单决策
"""
import json
import re
import pandas as pd
from typing import Dict, Any, Optional, List
from datetime import datetime
from app.utils.logger import logger
from app.services.llm_service import llm_service
from app.services.news_service import get_news_service
class MarketSignalAnalyzer:
"""市场信号分析器 - 只关注市场,输出客观信号"""
# 纯市场分析系统提示词(与旧版 CRYPTO_SYSTEM_PROMPT 保持一致,只移除仓位管理)
MARKET_ANALYSIS_PROMPT = """你是一位专业的加密货币交易员和技术分析师。你的任务是综合分析**K线数据、量价关系、技术指标和新闻舆情**,给出交易信号。
## 核心理念
加密货币市场波动大,每天都有交易机会。你的目标是:
- **主动寻找机会**,而不是被动等待完美信号
- 短线交易重点关注:超跌反弹、超涨回落、关键位突破
- 中线交易重点关注:趋势回调、形态突破、多周期共振
## 一、量价分析(最重要)
量价关系是判断趋势真假的核心:
### 1. 健康上涨信号
- **放量上涨**:价格上涨 + 成交量放大(量比>1.5= 上涨有效,可追多
- **缩量回调**:上涨后回调 + 成交量萎缩(量比<0.7= 回调健康,可低吸
### 2. 健康下跌信号
- **放量下跌**:价格下跌 + 成交量放大 = 下跌有效,可追空
- **缩量反弹**:下跌后反弹 + 成交量萎缩 = 反弹无力,可做空
### 3. 量价背离(重要反转信号)
- **顶背离**:价格创新高,但成交量未创新高 → 上涨动能衰竭,警惕回落
- **底背离**:价格创新低,但成交量未创新低 → 下跌动能衰竭,关注反弹
- **天量见顶**:极端放量(量比>3后价格滞涨 → 主力出货信号
- **地量见底**:极端缩量(量比<0.3)后价格企稳 → 抛压枯竭信号
### 4. 突破确认
- **有效突破**:突破关键位 + 放量确认(量比>1.5= 真突破
- **假突破**:突破关键位 + 缩量 = 假突破,可能回落
## 二、K线形态分析
### 反转形态
- **锤子线/倒锤子**:下跌趋势中出现,下影线长 = 底部信号
- **吞没形态**:大阳吞没前一根阴线 = 看涨;大阴吞没前一根阳线 = 看跌
- **十字星**:在高位/低位出现 = 变盘信号
- **早晨之星/黄昏之星**三根K线组合的反转信号
### 持续形态
- **三连阳/三连阴**:趋势延续信号
- **旗形整理**:趋势中的健康回调
## 三、技术指标分析
### RSI相对强弱指标
**RSI 是最重要的超买超卖指标,请注意细节:**
- **RSI < 30**:超卖区,关注反弹机会
- RSI 从 30 以下回升,交叉上穿 30买入信号
- RSI 底背离(价格新低但 RSI 未创新低):强买入信号
- **RSI > 70**:超买区,关注回落风险
- RSI 从 70 以上回落,交叉下穿 70卖出信号
- RSI 顶背离(价格新高但 RSI 未创新高):强卖出信号
- **RSI 40-60**:震荡区,观望为主
- **RSI 趋势**RSI 自身的趋势变化比单一数值更重要
### MACD
- 金叉DIF 上穿 DEA做多信号
- 死叉DIF 下穿 DEA做空信号
- 零轴上方金叉:强势做多
- 零轴下方死叉:强势做空
- MACD 柱状图背离:重要反转信号
### 布林带
- 触及下轨 + 企稳:反弹做多
- 触及上轨 + 受阻:回落做空
- 布林带收口:即将变盘
- 布林带开口:趋势启动
### 均线系统(重要)
**均线系统是趋势判断的核心,请仔细分析:**
- **多头排列**MA5 > MA10 > MA20 > MA50强势上涨趋势回调做多
- **空头排列**MA5 < MA10 < MA20 < MA50强势下跌趋势反弹做空
- **价格与 MA 的关系**
- 价格站稳 MA5/MA10 上方:短线上涨
- 价格突破 MA20中线转多
- 价格跌破 MA20中线转空
- MA50 是中期趋势的分水岭
- **均线金叉死叉**
- MA5 上穿 MA10短线买入信号
- MA5 下穿 MA10短线卖出信号
- MA10 上穿 MA20中线买入信号
- MA10 下穿 MA20中线卖出信号
## 四、新闻舆情分析
结合最新市场新闻判断:
- **重大利好**监管利好、机构入场、ETF 通过等 → 提高做多置信度
- **重大利空**:监管打压、交易所暴雷、黑客攻击等 → 提高做空置信度
- **市场情绪**:恐慌指数、社交媒体热度
- **大户动向**:鲸鱼转账、交易所流入流出
## 五、多周期共振(关键分析框架)
**多周期共振是提高信号质量的核心方法:**
### 周期层级关系
- **4h趋势层**:决定中期大方向
- **1h主周期**:主要交易周期
- **15m入场层**:寻找入场时机
- **5m精确入场**:确认最佳入场点
### 共振判断标准
**强共振A级信号**
- 所有周期趋势同向(如 4h多 + 1h多 + 15m多
- 多周期 RSI 同时超买/超卖后出现背离
- 多周期 MA 同时金叉/死叉
**中等共振B级信号**
- 大周期4h+1h同向
- 主周期1h技术指标明确
**弱共振C级信号**
- 只有单一周期信号
- 多周期方向不一致
### 实战策略
- **顺势交易**4h 和 1h 同向时,在 15m/5m 寻找入场点
- **逆势谨慎**:只有 1h 信号但 4h 反向时,降低置信度
- **突破交易**:多周期同时突破关键位,信号最强
## 六、入场方式
根据市场分析综合判断入场方式:
- **market**:现价立即入场
- 信号强烈且明确A级或高置信度B级
- 放量突破关键位,趋势明确
- 多周期共振,等待可能错过机会
- 市场波动大,等待可能价格变化太快
- **limit**:挂单等待入场
- 信号强度中等B级或C级
- 当前价格距离理想入场位有一定距离
- 判断市场可能回调到更好位置
- 希望获得更优成交价格,愿意承担可能无法成交的风险
**重要**
- 必须同时输出 `entry_zone`(建议入场价)和 `entry_type`(入场方式)
- 入场方式由你的市场分析判断,不是简单的价格距离计算
- 如果选择 `limit``entry_zone` 应该是你建议的挂单价格
## 输出格式
请严格按照以下 JSON 格式输出:
```json
{
"analysis_summary": "简要描述当前市场状态50字以内",
"volume_analysis": "量价分析结论30字以内",
"news_sentiment": "positive/negative/neutral",
"news_impact": "新闻对市场的影响分析30字以内",
"signals": [
{
"type": "short_term/medium_term/long_term",
"action": "buy/sell",
"entry_type": "market/limit",
"confidence": 0-100,
"grade": "A/B/C/D",
"entry_zone": 66000,
"stop_loss": 65500,
"take_profit": 67500,
"reasoning": "详细的入场理由(必须包含量价分析)",
"key_factors": ["关键因素1", "关键因素2"]
}
],
"key_levels": {
"support": [65000, 64500],
"resistance": [67000, 67500]
}
}
```
## 重要说明
- **所有价格必须是纯数字**,不要加 $ 符号、逗号或其他格式
- `entry_zone`、`stop_loss`、`take_profit` 必须是数字类型,不要是字符串
- `key_levels` 中的支撑位和阻力位也必须是数字数组
## 信号等级与置信度
- **A级**80-100量价配合 + 多指标共振 + 多周期确认
- **B级**60-79量价配合 + 主要指标确认
- **C级**40-59有机会但量价不够理想
- **D级**<40量价背离或信号矛盾
## 注意事项
1. **只在有明确的做多或做空机会时才输出信号**action 为 buy 或 sell
2. 如果市场不明朗,没有明确交易机会,**不要输出任何信号**signals 为空数组 []
3. 信号强度confidence要合理不要随意给高分
4. 60-70分一般信号可轻仓试探
5. 75-85分较强信号可正常仓位
6. 90+分:强信号,但也要控制风险
7. 不要输出 action 为 "wait" 的信号,如果没有交易机会就不输出
记住:你只负责分析市场,输出客观的交易信号,不需要考虑仓位管理和风险控制!
"""
def __init__(self):
self.news_service = get_news_service()
async def analyze(self, symbol: str, data: Dict[str, Any],
symbols: List[str] = None) -> Dict[str, Any]:
"""
分析市场并生成信号
Args:
symbol: 交易对
data: 多周期K线数据
symbols: 所有监控的交易对(用于市场对比)
Returns:
市场信号字典
"""
try:
# 1. 准备市场数据
market_context = self._prepare_market_context(symbol, data, symbols)
# 2. 获取新闻舆情
news_context = await self._get_news_context(symbol)
# 3. 构建 LLM 提示词
prompt = self._build_analysis_prompt(symbol, market_context, news_context)
# 4. 调用 LLM 分析
messages = [
{"role": "system", "content": self.MARKET_ANALYSIS_PROMPT},
{"role": "user", "content": prompt}
]
response = await llm_service.achat(messages)
# 5. 解析结果
result = self._parse_llm_response(response, symbol)
return result
except Exception as e:
logger.error(f"市场信号分析失败: {e}")
import traceback
logger.debug(traceback.format_exc())
return self._get_empty_signal(symbol)
def _prepare_market_context(self, symbol: str, data: Dict,
symbols: List[str] = None) -> str:
"""准备市场上下文信息"""
context_parts = []
# 当前价格和24h变化
current_price = float(data['5m'].iloc[-1]['close'])
price_change_24h = self._calculate_price_change_24h(data['1h'])
context_parts.append(f"当前价格: ${current_price:,.2f} ({price_change_24h})")
# 多周期数据
for tf_name, df in data.items():
if df is None or len(df) == 0:
continue
latest = df.iloc[-1]
context_parts.append(f"\n## {tf_name} 数据")
context_parts.append(f"开: {latest['open']}, 高: {latest['high']}, 低: {latest['low']}, 收: {latest['close']}")
context_parts.append(f"成交量: {latest.get('volume', 'N/A')}")
# 技术指标
if 'rsi' in df.columns:
rsi = df['rsi'].iloc[-1]
context_parts.append(f"RSI: {rsi:.2f}")
if 'macd' in df.columns:
macd = df['macd'].iloc[-1]
signal = df['macd_signal'].iloc[-1]
context_parts.append(f"MACD: {macd:.4f}, 信号线: {signal:.4f}")
if 'bb_upper' in df.columns:
bb_upper = df['bb_upper'].iloc[-1]
bb_lower = df['bb_lower'].iloc[-1]
context_parts.append(f"布林带: 上轨 {bb_upper:.2f}, 下轨 {bb_lower:.2f}")
# 均线系统
context_parts.append(f"\n## 均线系统")
df_1h = data.get('1h')
if df_1h is not None and len(df_1h) > 0:
latest = df_1h.iloc[-1]
context_parts.append(f"MA5: {latest.get('ma5', 'N/A')}")
context_parts.append(f"MA10: {latest.get('ma10', 'N/A')}")
context_parts.append(f"MA20: {latest.get('ma20', 'N/A')}")
context_parts.append(f"MA50: {latest.get('ma50', 'N/A')}")
# 判断均线排列
ma5 = latest.get('ma5', 0)
ma10 = latest.get('ma10', 0)
ma20 = latest.get('ma20', 0)
ma50 = latest.get('ma50', 0)
if all([ma5, ma10, ma20, ma50]):
if ma5 > ma10 > ma20 > ma50:
context_parts.append("均线排列: 多头排列 📈")
elif ma5 < ma10 < ma20 < ma50:
context_parts.append("均线排列: 空头排列 📉")
else:
context_parts.append("均线排列: 交织,方向不明")
# 量比分析
df_5m = data.get('5m')
if df_5m is not None and len(df_5m) >= 20:
vol_latest = df_5m['volume'].iloc[-1]
vol_ma20 = df_5m['volume'].iloc[-20:-1].mean()
volume_ratio = vol_latest / vol_ma20 if vol_ma20 > 0 else 1
context_parts.append(f"\n## 量价分析")
context_parts.append(f"最新成交量: {vol_latest:.0f}")
context_parts.append(f"20周期均量: {vol_ma20:.0f}")
context_parts.append(f"量比: {volume_ratio:.2f}")
if volume_ratio > 1.5:
context_parts.append("量价状态: 放量 📊")
elif volume_ratio < 0.7:
context_parts.append("量价状态: 缩量 📉")
else:
context_parts.append("量价状态: 平量 ")
# 波动率分析
volatility_analysis = self._analyze_volatility(data)
if volatility_analysis:
context_parts.append(f"\n## 波动率分析")
context_parts.append(volatility_analysis)
return "\n".join(context_parts)
async def _get_news_context(self, symbol: str) -> str:
"""获取新闻舆情上下文"""
try:
news_result = await self.news_service.get_crypto_news(symbol)
if not news_result or not news_result.get('articles'):
return "无最新新闻"
articles = news_result['articles'][:5] # 只取前5条
context_parts = ["\n## 最新新闻"]
for article in articles:
title = article.get('title', '')
source = article.get('source', '')
published_at = article.get('publishedAt', '')
time_str = published_at.split('T')[1][:5] if 'T' in published_at else ''
context_parts.append(f"- [{time_str}] {title} ({source})")
return "\n".join(context_parts)
except Exception as e:
logger.warning(f"获取新闻失败: {e}")
return "新闻获取失败"
def _build_analysis_prompt(self, symbol: str, market_context: str,
news_context: str) -> str:
"""构建分析提示词"""
return f"""请分析 {symbol} 的市场情况:
{market_context}
{news_context}
请根据以上数据,给出你的市场判断和交易信号。
"""
def _parse_llm_response(self, response: str, symbol: str) -> Dict[str, Any]:
"""解析 LLM 响应"""
try:
# 尝试提取 JSON
json_match = re.search(r'```json\s*([\s\S]*?)\s*```', response)
if json_match:
json_str = json_match.group(1)
else:
json_match = re.search(r'\{[\s\S]*\}', response)
if json_match:
json_str = json_match.group(0)
else:
raise ValueError("无法找到 JSON 响应")
# 清理 JSON 字符串(移除可能导致解析错误的注释等)
json_str = self._clean_json_string(json_str)
logger.debug(f"解析的 JSON 字符串: {json_str[:500]}...") # 打印前500字符用于调试
result = json.loads(json_str)
# 清理价格字段 - 转换为 float
result = self._clean_price_fields(result)
# 添加元数据
result['symbol'] = symbol
result['timestamp'] = datetime.now().isoformat()
result['raw_response'] = response
# 兼容处理:确保 signals 中的字段与旧格式一致
if 'signals' in result:
for sig in result['signals']:
# LLM 输出的 "type" 是 timeframe (short_term/medium_term/long_term)
# 需要映射为 "timeframe",而 "action" 才是 buy/sell/wait
if 'type' in sig:
# 如果 type 是 short_term/medium_term/long_term映射为 timeframe
if sig['type'] in ['short_term', 'medium_term', 'long_term']:
sig['timeframe'] = sig.pop('type')
# 如果 type 是 buy/sell/wait映射为 action
elif sig['type'] in ['buy', 'sell', 'wait']:
sig['action'] = sig.pop('type')
# 确保 action 字段存在
if 'action' not in sig and 'timeframe' in sig:
# 从 reasoning 或其他字段推断 action
sig['action'] = 'wait'
# 确保 grade 字段存在
if 'grade' not in sig:
# 根据 confidence 推断 grade
confidence = sig.get('confidence', 0)
if confidence >= 80:
sig['grade'] = 'A'
elif confidence >= 60:
sig['grade'] = 'B'
elif confidence >= 40:
sig['grade'] = 'C'
else:
sig['grade'] = 'D'
# 从信号中推断 market_state 和 trend
if 'signals' in result and result['signals']:
# 找出置信度最高的信号
best_signal = max(result['signals'], key=lambda s: s.get('confidence', 0))
action = best_signal.get('action', 'wait')
confidence = best_signal.get('confidence', 0)
# 推断市场状态
if confidence >= 70:
if action == 'buy':
result['market_state'] = '强势上涨'
elif action == 'sell':
result['market_state'] = '强势下跌'
else:
result['market_state'] = '震荡整理'
else:
result['market_state'] = '震荡整理'
# 推断趋势
if action == 'buy':
result['trend'] = 'up'
elif action == 'sell':
result['trend'] = 'down'
else:
result['trend'] = 'sideways'
else:
result['market_state'] = '无明确信号'
result['trend'] = 'sideways'
logger.info(f"✅ 市场信号分析完成: {symbol}")
logger.debug(f"市场信号: {json.dumps(result, ensure_ascii=False, indent=2)}")
return result
except Exception as e:
logger.warning(f"解析 LLM 响应失败: {e}")
logger.warning(f"原始响应: {response[:1000]}...") # 打印前1000字符
return self._get_empty_signal(symbol)
def _clean_json_string(self, json_str: str) -> str:
"""清理 JSON 字符串,移除可能导致解析错误的内容"""
# 移除单行注释 // ...
json_str = re.sub(r'//.*?(?=\n|$)', '', json_str)
# 移除多行注释 /* ... */
json_str = re.sub(r'/\*[\s\S]*?\*/', '', json_str)
# 移除尾随逗号(例如 {"a": 1,} -> {"a": 1}
json_str = re.sub(r',\s*([}\]])', r'\1', json_str)
return json_str
def _clean_price_fields(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""清理价格字段,转换为 float"""
def clean_price(price_value):
if price_value is None:
return None
if isinstance(price_value, (int, float)):
return float(price_value)
if isinstance(price_value, str):
# 移除 $ 符号和逗号
cleaned = price_value.replace('$', '').replace(',', '').strip()
if cleaned:
try:
return float(cleaned)
except ValueError:
return None
return None
# 清理 key_levels 中的支撑位和阻力位
if 'key_levels' in data and data['key_levels']:
key_levels = data['key_levels']
if 'support' in key_levels:
data['key_levels']['support'] = [clean_price(s) for s in key_levels['support']]
if 'resistance' in key_levels:
data['key_levels']['resistance'] = [clean_price(r) for r in key_levels['resistance']]
# 清理 signals 中的价格字段
if 'signals' in data:
for sig in data['signals']:
price_fields = ['entry_zone', 'stop_loss', 'take_profit']
for field in price_fields:
if field in sig:
sig[field] = clean_price(sig[field])
return data
def _calculate_price_change_24h(self, df) -> str:
"""计算24小时涨跌幅"""
try:
if df is None or len(df) < 24:
return "N/A"
current_price = float(df['close'].iloc[-1])
price_24h_ago = float(df['close'].iloc[-24])
change = ((current_price - price_24h_ago) / price_24h_ago) * 100
sign = "+" if change >= 0 else ""
return f"{sign}{change:.2f}%"
except Exception as e:
logger.debug(f"计算24h涨跌失败: {e}")
return "N/A"
def _get_empty_signal(self, symbol: str) -> Dict[str, Any]:
"""返回空信号"""
return {
'symbol': symbol,
'analysis_summary': 'unknown',
'volume_analysis': '分析失败',
'news_sentiment': 'neutral',
'news_impact': '',
'market_state': '分析失败',
'trend': 'sideways',
'signals': [],
'key_levels': {},
'timestamp': datetime.now().isoformat(),
'error': '信号分析失败'
}
def _analyze_volatility(self, data: Dict[str, pd.DataFrame]) -> str:
"""分析波动率变化"""
df = data.get('1h')
if df is None or len(df) < 24 or 'atr' not in df.columns:
return ""
lines = []
# ATR 变化趋势
recent_atr = df['atr'].iloc[-6:].mean() # 最近 6 根
older_atr = df['atr'].iloc[-12:-6].mean() # 之前 6 根
if pd.isna(recent_atr) or pd.isna(older_atr) or older_atr == 0:
return ""
atr_change = (recent_atr - older_atr) / older_atr * 100
current_atr = float(df['atr'].iloc[-1])
current_price = float(df['close'].iloc[-1])
atr_percent = current_atr / current_price * 100
lines.append(f"当前 ATR: ${current_atr:.2f} ({atr_percent:.2f}%)")
if atr_change > 20:
lines.append(f"**波动率扩张**: ATR 上升 {atr_change:.0f}%,趋势可能启动")
elif atr_change < -20:
lines.append(f"**波动率收缩**: ATR 下降 {abs(atr_change):.0f}%,可能即将突破")
else:
lines.append(f"波动率稳定: ATR 变化 {atr_change:+.0f}%")
# 布林带宽度
if 'bb_upper' in df.columns and 'bb_lower' in df.columns:
bb_width = (float(df['bb_upper'].iloc[-1]) - float(df['bb_lower'].iloc[-1])) / current_price * 100
bb_width_prev = (float(df['bb_upper'].iloc[-6]) - float(df['bb_lower'].iloc[-6])) / float(df['close'].iloc[-6]) * 100
if bb_width < bb_width_prev * 0.8:
lines.append(f"**布林带收口**: 宽度 {bb_width:.1f}%,变盘信号")
elif bb_width > bb_width_prev * 1.2:
lines.append(f"**布林带开口**: 宽度 {bb_width:.1f}%,趋势延续")
return "\n".join(lines)