aaron/tradusai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
aaron 2025-12-11 22:51:51 +08:00
parent b2f994e2e9
commit 73bb91b6f2
3 changed files with 697 additions and 308 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

157
notifiers/dingtalk.py
View File

@ -194,6 +194,10 @@ class DingTalkNotifier:
"""
格式化交易信号为Markdown文本（多时间级别版本）
支持两种格式:
1. 新格式: trades数组 (优先)
2. 旧格式: opportunities对象 (向后兼容)
Args:
signal: 聚合信号
@ -214,7 +218,8 @@ class DingTalkNotifier:
lines = []
# === 核心信号 ===
lines.append(f"# {emoji} {signal_type}")
symbol = signal.get('symbol', 'BTC/USDT')
lines.append(f"# {emoji} {symbol} {signal_type}")
lines.append("")
lines.append(f"**综合置信度**: {confidence:.0%} | **时间**: {datetime.now().strftime('%H:%M')}")
lines.append("")
@ -233,6 +238,45 @@ class DingTalkNotifier:
# 获取LLM信号
llm_signal = signal.get('llm_signal') or {}
# 检测是否为新格式 (trades数组)
trades = llm_signal.get('trades', [])
if trades and isinstance(trades, list) and len(trades) >= 3:
# 新格式: trades数组
trades_by_tf = {t.get('timeframe'): t for t in trades if t.get('timeframe')}
# 短期分析
self._add_trade_section(
lines,
"短期 (5m/15m/1h)",
"⚡",
trades_by_tf.get('short', {}),
signal
)
# 中期分析
self._add_trade_section(
lines,
"中期 (4h/1d)",
"📈",
trades_by_tf.get('medium', {}),
signal
)
# 长期分析
self._add_trade_section(
lines,
"长期 (1d/1w)",
"📅",
trades_by_tf.get('long', {}),
signal
)
# 综合分析
analysis = llm_signal.get('analysis', {})
reason = analysis.get('summary', '') or llm_signal.get('reasoning', '')
else:
# 旧格式: opportunities对象
opportunities = llm_signal.get('opportunities', {})
recommendations = llm_signal.get('recommendations_by_timeframe', {})
@ -243,7 +287,7 @@ class DingTalkNotifier:
"⚡",
opportunities.get('short_term_5m_15m_1h', {}),
recommendations.get('short_term', ''),
signal # 传递完整信号数据
signal
)
# 中期分析
@ -266,8 +310,9 @@ class DingTalkNotifier:
signal
)
# === 综合建议 ===
reason = llm_signal.get('reasoning', '') or self._get_brief_reason(signal)
# === 综合建议 ===
if reason:
lines.append("---")
lines.append("## 💡 综合分析")
@ -275,12 +320,118 @@ class DingTalkNotifier:
lines.append(f"{reason}")
lines.append("")
# === 关键价位 ===
key_levels = llm_signal.get('key_levels', {})
if key_levels:
support = key_levels.get('support', [])
resistance = key_levels.get('resistance', [])
if support or resistance:
lines.append("---")
lines.append("## 📍 关键价位")
lines.append("")
if support:
support_str = ", ".join([f"${p:,.0f}" for p in support[:3]])
lines.append(f"**支撑**: {support_str}")
if resistance:
resistance_str = ", ".join([f"${p:,.0f}" for p in resistance[:3]])
lines.append(f"**阻力**: {resistance_str}")
lines.append("")
# === 页脚 ===
lines.append("---")
lines.append("*仅供参考，不构成投资建议*")
return "\n".join(lines)
def _add_trade_section(
self,
lines: list,
timeframe_label: str,
emoji: str,
trade: Dict[str, Any],
signal: Dict[str, Any] = None
):
"""
添加单个时间级别的交易区块（新格式trades数组）
Args:
lines: 输出行列表
timeframe_label: 时间级别标签
emoji: emoji图标
trade: 该时间级别的交易信息
signal: 完整信号数据
"""
lines.append(f"### {emoji} {timeframe_label}")
lines.append("")
status = trade.get('status', 'INACTIVE')
is_active = status == 'ACTIVE'
if is_active:
direction = trade.get('direction', 'NONE')
entry = trade.get('entry', {})
exit_data = trade.get('exit', {})
position = trade.get('position', {})
risk_reward = trade.get('risk_reward', 0)
expected_profit = trade.get('expected_profit_pct', 0)
reasoning = trade.get('reasoning', '')
# 方向标识
direction_emoji = "🟢" if direction == "LONG" else "🔴" if direction == "SHORT" else "⚪"
lines.append(f"{direction_emoji} **方向**: {direction}")
lines.append("")
# 金字塔入场价格
entry_prices = []
for i in range(1, 5):
price = entry.get(f'price_{i}', 0)
pct = position.get(f'size_pct_{i}', 0)
if price > 0:
entry_prices.append(f"${price:,.0f}({pct}%)")
if entry_prices:
lines.append(f"**入场**: {' → '.join(entry_prices)}")
# 止损止盈
stop_loss = exit_data.get('stop_loss', 0)
tp1 = exit_data.get('take_profit_1', 0)
tp2 = exit_data.get('take_profit_2', 0)
tp3 = exit_data.get('take_profit_3', 0)
if stop_loss > 0:
lines.append(f"**止损**: ${stop_loss:,.0f}")
take_profits = []
if tp1 > 0:
take_profits.append(f"${tp1:,.0f}")
if tp2 > 0:
take_profits.append(f"${tp2:,.0f}")
if tp3 > 0:
take_profits.append(f"${tp3:,.0f}")
if take_profits:
lines.append(f"**止盈**: {' / '.join(take_profits)}")
# 风险回报比和预期盈利
if risk_reward > 0:
lines.append(f"**风险回报**: 1:{risk_reward:.1f}")
if expected_profit > 0:
lines.append(f"**预期盈利**: {expected_profit:.1f}%")
lines.append("")
# 理由
if reasoning:
lines.append(f"💭 {reasoning}")
lines.append("")
else:
# 无交易机会
reasoning = trade.get('reasoning', '')
if reasoning:
lines.append(f"💭 {reasoning}")
else:
lines.append("💭 暂无明确交易机会")
lines.append("")
def _add_timeframe_section(
self,
lines: list,

622
signals/llm_decision.py
View File

@ -146,111 +146,151 @@ class LLMDecisionMaker:
current_price = market_context.get('current_price', 0)
kline_data = market_context.get('kline_data', {})
# Build structured prompt
prompt = f"""你是一个专业的加密货币交易分析师。基于以下多时间周期的K线数据和技术指标,提供分层次的交易建议。
# Build structured prompt - 自动化交易友好的JSON格式
prompt = f"""你是一个专业的加密货币交易分析师。基于以下多时间周期的K线数据和技术指标,提供精确的交易信号。
**重要**: 你需要自己从K线数据中识别支撑位和压力位,不要依赖预先计算的值。
## 当前价格: ${current_price:,.2f}
## 当前价格
${current_price:,.2f}
## 输出要求
## 你的分析任务
请严格按照以下JSON Schema输出，所有字段必须存在，价格字段必须是数字(无机会时用0):
1. **分析K线数据** - 识别各周期的支撑位、压力位、趋势结构
2. **结合技术指标** - RSI、MACD、成交量等确认信号
3. **给出交易建议** - 分短期/中期/长期三个级别
## 请提供以下内容 (使用JSON格式):
```json
{{
"signal": "BUY" | "SELL" | "HOLD",
"confidence": 0.0-1.0,
"signal": "BUY|SELL|HOLD",
"confidence": 0.65,
"risk_level": "LOW|MEDIUM|HIGH",
"market_bias": "BULLISH|BEARISH|NEUTRAL",
"trades": [
{{
"id": "short_001",
"timeframe": "short",
"status": "ACTIVE|INACTIVE",
"direction": "LONG|SHORT|NONE",
"entry": {{
"price_1": 90000.00,
"price_2": 89700.00,
"price_3": 89400.00,
"price_4": 89100.00
}},
"exit": {{
"stop_loss": 88500.00,
"take_profit_1": 91000.00,
"take_profit_2": 92000.00,
"take_profit_3": 93000.00
}},
"position": {{
"size_pct_1": 40,
"size_pct_2": 30,
"size_pct_3": 20,
"size_pct_4": 10
}},
"risk_reward": 2.5,
"expected_profit_pct": 1.5,
"reasoning": "简要说明"
}},
{{
"id": "medium_001",
"timeframe": "medium",
"status": "ACTIVE|INACTIVE",
"direction": "LONG|SHORT|NONE",
"entry": {{
"price_1": 89500.00,
"price_2": 89000.00,
"price_3": 88500.00,
"price_4": 88000.00
}},
"exit": {{
"stop_loss": 86000.00,
"take_profit_1": 93000.00,
"take_profit_2": 95000.00,
"take_profit_3": 98000.00
}},
"position": {{
"size_pct_1": 40,
"size_pct_2": 30,
"size_pct_3": 20,
"size_pct_4": 10
}},
"risk_reward": 2.0,
"expected_profit_pct": 3.5,
"reasoning": "简要说明"
}},
{{
"id": "long_001",
"timeframe": "long",
"status": "ACTIVE|INACTIVE",
"direction": "LONG|SHORT|NONE",
"entry": {{
"price_1": 88000.00,
"price_2": 86000.00,
"price_3": 84000.00,
"price_4": 82000.00
}},
"exit": {{
"stop_loss": 78000.00,
"take_profit_1": 95000.00,
"take_profit_2": 100000.00,
"take_profit_3": 110000.00
}},
"position": {{
"size_pct_1": 40,
"size_pct_2": 30,
"size_pct_3": 20,
"size_pct_4": 10
}},
"risk_reward": 2.0,
"expected_profit_pct": 8.0,
"reasoning": "简要说明"
}}
],
// 你识别的关键价位
"key_levels": {{
"short_term": {{
"support": [支撑位数组],
"resistance": [压力位数组]
}},
"medium_term": {{
"support": [支撑位数组],
"resistance": [压力位数组]
}},
"long_term": {{
"support": [支撑位数组],
"resistance": [压力位数组]
}}
"support": [89000.00, 88000.00, 86000.00],
"resistance": [92000.00, 94000.00, 96000.00]
}},
// 分时间级别的交易机会分析 - 支持金字塔加仓的多级进场
"opportunities": {{
"short_term_5m_15m_1h": {{
"exists": true/false,
"direction": "LONG" | "SHORT" | null,
"entry_levels": [
{{"price": 首次进场价格, "ratio": 0.4, "reasoning": "首仓理由"}},
{{"price": 第二次进场价格, "ratio": 0.3, "reasoning": "加仓1理由"}},
{{"price": 第三次进场价格, "ratio": 0.2, "reasoning": "加仓2理由"}},
{{"price": 第四次进场价格, "ratio": 0.1, "reasoning": "加仓3理由"}}
],
"stop_loss": 止损价格数值或null,
"take_profit": 止盈价格数值或null,
"reasoning": "短期日内机会说明"
}},
"medium_term_4h_1d": {{
"exists": true/false,
"direction": "LONG" | "SHORT" | null,
"entry_levels": [
{{"price": 首次进场价格, "ratio": 0.4, "reasoning": "首仓理由"}},
{{"price": 第二次进场价格, "ratio": 0.3, "reasoning": "加仓1理由"}},
{{"price": 第三次进场价格, "ratio": 0.2, "reasoning": "加仓2理由"}},
{{"price": 第四次进场价格, "ratio": 0.1, "reasoning": "加仓3理由"}}
],
"stop_loss": 止损价格数值或null,
"take_profit": 止盈价格数值或null,
"reasoning": "中期波段机会说明"
}},
"long_term_1d_1w": {{
"exists": true/false,
"direction": "LONG" | "SHORT" | null,
"entry_levels": [
{{"price": 首次进场价格, "ratio": 0.4, "reasoning": "首仓理由"}},
{{"price": 第二次进场价格, "ratio": 0.3, "reasoning": "加仓1理由"}},
{{"price": 第三次进场价格, "ratio": 0.2, "reasoning": "加仓2理由"}},
{{"price": 第四次进场价格, "ratio": 0.1, "reasoning": "加仓3理由"}}
],
"stop_loss": 止损价格数值或null,
"take_profit": 止盈价格数值或null,
"reasoning": "长期趋势机会说明"
}},
"ambush": {{
"exists": true/false,
"price_level": 埋伏价格数值或null,
"reasoning": "埋伏点位说明"
}}
"analysis": {{
"trend": "UP|DOWN|SIDEWAYS",
"momentum": "STRONG|WEAK|NEUTRAL",
"volume": "HIGH|LOW|NORMAL",
"summary": "一句话市场总结"
}},
// 分级别操作建议（必填）
"recommendations_by_timeframe": {{
"short_term": "短期(5m/15m/1h)操作建议",
"medium_term": "中期(4h/1d)操作建议",
"long_term": "长期(1d/1w)操作建议"
}},
"reasoning": "多周期综合分析",
"risk_level": "LOW" | "MEDIUM" | "HIGH",
"key_factors": ["影响因素1", "影响因素2", ...]
"key_factors": ["因素1", "因素2", "因素3"]
}}
```
**重要原则**:
1. **优先日内短线** - 重点关注 short_term_5m_15m_1h 的日内交易机会
2. **不同周期盈利要求不同** - 短期≥1%，中期≥2%，长期≥5%，不满足则 exists=false
3. **自行识别支撑压力位** - 从K线数据中找出重要的高低点作为支撑压力位
4. **响应必须是有效的JSON格式** - 不要包含注释
5. **金字塔加仓策略** - entry_levels 必须包含4个价位:
- 做多: 首仓价格最高，后续价位逐渐降低 (越跌越买)
- 做空: 首仓价格最低，后续价位逐渐升高 (越涨越卖)
- ratio总和=1.0 (0.4+0.3+0.2+0.1)
- 各级价位间距建议: 短期0.3-0.5%，中期0.5-1%，长期1-2%
## 字段说明
**trades数组** (必须包含3个元素，分别对应short/medium/long):
- `timeframe`: "short"=5m/15m/1h, "medium"=4h/1d, "long"=1d/1w
- `status`: "ACTIVE"=有交易机会, "INACTIVE"=暂无机会
- `direction`: "LONG"=做多, "SHORT"=做空, "NONE"=无方向
- `entry.price_1~4`: 金字塔4级进场价，做多时price_1最高逐渐降低，做空时price_1最低逐渐升高
- `exit.stop_loss`: 统一止损价
- `exit.take_profit_1~3`: 3级止盈目标
- `position.size_pct_1~4`: 各级仓位百分比，总和=100
- `risk_reward`: 风险回报比
- `expected_profit_pct`: 预期盈利百分比
**盈利要求** (不满足时status=INACTIVE):
- short: expected_profit_pct >= 1.0%
- medium: expected_profit_pct >= 2.0%
- long: expected_profit_pct >= 5.0%
**价格间距建议**:
- short: 各级入场价间距 0.3-0.5%
- medium: 各级入场价间距 0.5-1.0%
- long: 各级入场价间距 1.0-2.0%
## 重要规则
1. **所有价格字段必须是数字**，无机会时填0，不要用null
2. **trades数组必须有3个元素**，顺序: short, medium, long
3. **status=INACTIVE时**，direction设为"NONE"，所有价格设为0
4. **只输出JSON**，不要有其他文字
"""
@ -319,33 +359,23 @@ ${current_price:,.2f}
# Add analysis guidelines
prompt += """
## 支撑压力位识别方法
## 分析指南
1. **短期支撑压力 (5m/15m/1h)**
- 近1天内的明显高低点
- 多次触及但未突破的价格
- 整数关口 (如 91000, 92000)
### 支撑压力位识别
1. **短期**: 近1天内明显高低点、整数关口
2. **中期**: 近几天重要高低点、趋势线
3. **长期**: 周线/月线级别高低点
2. **中期支撑压力 (4h/1d)**
- 近几天的重要高低点
- 趋势线位置
- 前期成交密集区
### 止盈止损设置
- **short**: 止损0.3-0.5%, 止盈≥1%
- **medium**: 止损1-2%, 止盈≥2%
- **long**: 止损2-4%, 止盈≥5%
3. **长期支撑压力 (1d/1w)**
- 周线/月线级别的高低点
- 历史重要价格区间
- 大周期趋势线
## 止盈止损设置（不同周期要求不同）
- 短期 (5m/15m/1h): 止损 0.3%-0.5%, 止盈 ≥1%
- 中期 (4h/1d): 止损 1%-2%, 止盈 ≥2%
- 长期 (1d/1w): 止损 2%-4%, 止盈 ≥5%
重要：各周期的盈利空间必须满足最低要求才给出建议：
- 短期机会: (take_profit - entry) / entry ≥ 1%
- 中期机会: (take_profit - entry) / entry ≥ 2%
- 长期机会: (take_profit - entry) / entry ≥ 5%
### 最终检查
1. 确保JSON格式正确，无注释
2. 确保trades数组有3个元素
3. 确保所有价格是数字不是null
4. 确保INACTIVE时所有价格为0
"""
return prompt
@ -400,7 +430,7 @@ ${current_price:,.2f}
response_text: str,
market_context: Dict[str, Any]
) -> Dict[str, Any]:
"""Parse LLM response into structured decision"""
"""Parse LLM response into structured decision - 支持新版trades数组格式"""
# Try to extract JSON from response
json_match = re.search(r'\{[\s\S]*\}', response_text)
@ -419,7 +449,6 @@ ${current_price:,.2f}
# Helper function to safely convert to float
def safe_float(value, default=0.0):
"""Safely convert value to float, handling None and invalid values"""
if value is None:
return default
try:
@ -427,9 +456,165 @@ ${current_price:,.2f}
except (ValueError, TypeError):
return default
# Helper function to calculate profit percentage
current_price = market_context.get('current_price', 0)
# ========== 检测新版trades数组格式 ==========
trades = llm_decision.get('trades', [])
if trades and isinstance(trades, list) and len(trades) >= 3:
# 新版格式：trades数组
return self._parse_new_format(llm_decision, market_context, safe_float)
else:
# 旧版格式：opportunities对象
return self._parse_old_format(llm_decision, market_context, safe_float)
def _parse_new_format(
self,
llm_decision: Dict[str, Any],
market_context: Dict[str, Any],
safe_float
) -> Dict[str, Any]:
"""解析新版trades数组格式"""
current_price = market_context.get('current_price', 0)
trades = llm_decision.get('trades', [])
# 构建trades字典 (by timeframe)
trades_by_tf = {}
for trade in trades:
tf = trade.get('timeframe', '')
if tf in ['short', 'medium', 'long']:
trades_by_tf[tf] = trade
# 转换为统一的opportunities格式 (向后兼容)
def convert_trade_to_opportunity(trade: Dict, tf_key: str) -> Dict:
"""将新格式trade转换为旧格式opportunity"""
if not trade or trade.get('status') != 'ACTIVE':
return {
'exists': False,
'direction': None,
'entry_levels': [],
'entry_price': 0,
'stop_loss': 0,
'take_profit': 0,
'reasoning': trade.get('reasoning', '') if trade else ''
}
entry = trade.get('entry', {})
exit_data = trade.get('exit', {})
position = trade.get('position', {})
# 构建entry_levels
entry_levels = []
for i in range(1, 5):
price = safe_float(entry.get(f'price_{i}'), 0)
ratio = safe_float(position.get(f'size_pct_{i}'), [40, 30, 20, 10][i-1]) / 100
if price > 0:
entry_levels.append({
'price': price,
'ratio': ratio,
'level': i - 1,
'reasoning': ''
})
return {
'exists': True,
'direction': trade.get('direction', 'NONE'),
'entry_levels': entry_levels,
'entry_price': safe_float(entry.get('price_1'), 0),
'stop_loss': safe_float(exit_data.get('stop_loss'), 0),
'take_profit': safe_float(exit_data.get('take_profit_1'), 0),
'take_profit_2': safe_float(exit_data.get('take_profit_2'), 0),
'take_profit_3': safe_float(exit_data.get('take_profit_3'), 0),
'risk_reward': safe_float(trade.get('risk_reward'), 0),
'expected_profit_pct': safe_float(trade.get('expected_profit_pct'), 0),
'reasoning': trade.get('reasoning', '')
}
short_opp = convert_trade_to_opportunity(trades_by_tf.get('short'), 'short')
medium_opp = convert_trade_to_opportunity(trades_by_tf.get('medium'), 'medium')
long_opp = convert_trade_to_opportunity(trades_by_tf.get('long'), 'long')
# 获取分析数据
analysis = llm_decision.get('analysis', {})
key_levels = llm_decision.get('key_levels', {})
# 构建最终decision
decision = {
'timestamp': datetime.now().isoformat(),
'signal_type': llm_decision.get('signal', 'HOLD').upper(),
'confidence': safe_float(llm_decision.get('confidence'), 0.5),
'trade_type': 'MULTI_TIMEFRAME',
'market_bias': llm_decision.get('market_bias', 'NEUTRAL'),
'risk_level': llm_decision.get('risk_level', 'MEDIUM'),
# 新版trades数组 (原始格式保留)
'trades': trades,
# 向后兼容的opportunities格式
'opportunities': {
'short_term_5m_15m_1h': short_opp,
'medium_term_4h_1d': medium_opp,
'long_term_1d_1w': long_opp,
# 向后兼容
'intraday': short_opp,
'swing': medium_opp if medium_opp.get('exists') else long_opp,
},
# 分析数据
'analysis': {
'trend': analysis.get('trend', 'SIDEWAYS'),
'momentum': analysis.get('momentum', 'NEUTRAL'),
'volume': analysis.get('volume', 'NORMAL'),
'summary': analysis.get('summary', ''),
},
# 关键价位
'key_levels': {
'support': key_levels.get('support', []),
'resistance': key_levels.get('resistance', []),
},
'key_factors': llm_decision.get('key_factors', []),
'reasoning': analysis.get('summary', ''),
# 价格水平 (向后兼容)
'levels': {
'current_price': current_price,
'entry': short_opp['entry_price'] or medium_opp['entry_price'] or long_opp['entry_price'],
'stop_loss': short_opp['stop_loss'] or medium_opp['stop_loss'] or long_opp['stop_loss'],
'take_profit_1': short_opp['take_profit'] or medium_opp['take_profit'] or long_opp['take_profit'],
'take_profit_2': short_opp.get('take_profit_2', 0) or medium_opp.get('take_profit_2', 0),
'take_profit_3': short_opp.get('take_profit_3', 0) or medium_opp.get('take_profit_3', 0),
},
'raw_response': '', # Will be set by caller
}
# Calculate risk-reward ratio
entry = decision['levels']['entry']
stop_loss = decision['levels']['stop_loss']
tp1 = decision['levels']['take_profit_1']
if entry and stop_loss and tp1 and entry != stop_loss:
risk = abs(entry - stop_loss)
reward = abs(tp1 - entry)
decision['risk_reward_ratio'] = round(reward / risk, 2) if risk > 0 else 0
else:
decision['risk_reward_ratio'] = 0
return decision
def _parse_old_format(
self,
llm_decision: Dict[str, Any],
market_context: Dict[str, Any],
safe_float
) -> Dict[str, Any]:
"""解析旧版opportunities格式 (向后兼容)"""
current_price = market_context.get('current_price', 0)
# Helper functions
def calc_profit_pct(entry, take_profit, direction):
"""Calculate profit percentage for a trade"""
if not entry or not take_profit or entry <= 0:
return 0
if direction == 'LONG':
@ -438,9 +623,7 @@ ${current_price:,.2f}
return (entry - take_profit) / entry * 100
return 0
# Helper function to check if opportunity meets minimum profit threshold
def meets_profit_threshold(opp, min_profit_pct=1.0):
"""Check if opportunity has at least min_profit_pct profit potential"""
if not opp.get('exists'):
return False
entry = safe_float(opp.get('entry_price'), 0)
@ -449,172 +632,76 @@ ${current_price:,.2f}
profit_pct = calc_profit_pct(entry, tp, direction)
return profit_pct >= min_profit_pct
# Parse opportunities structure (support both old and new format)
opportunities = llm_decision.get('opportunities', {})
# Helper function to normalize entry_levels
def normalize_entry_levels(opp: dict, direction: str, current_price: float) -> list:
"""Normalize entry_levels format, handling both new and old formats"""
def normalize_entry_levels(opp: dict, direction: str) -> list:
entry_levels = opp.get('entry_levels', [])
if entry_levels and isinstance(entry_levels, list):
# New format with entry_levels array
normalized = []
for i, level in enumerate(entry_levels[:4]): # Max 4 levels
for i, level in enumerate(entry_levels[:4]):
if isinstance(level, dict):
normalized.append({
'price': safe_float(level.get('price'), 0),
'ratio': safe_float(level.get('ratio'), [0.4, 0.3, 0.2, 0.1][i]),
'reasoning': level.get('reasoning', ''),
'level': i,
})
elif isinstance(level, (int, float)):
normalized.append({
'price': safe_float(level, 0),
'ratio': [0.4, 0.3, 0.2, 0.1][i],
'reasoning': '',
'level': i,
})
return normalized
# Fallback: convert old single entry_price format to entry_levels
entry_price = safe_float(opp.get('entry_price'), 0)
if entry_price <= 0:
entry_price = current_price
# Generate 4 levels with default spacing
entry_price = safe_float(opp.get('entry_price'), current_price)
levels = []
if direction == 'LONG':
# For LONG: first entry highest, subsequent entries lower
spacings = [0, 0.003, 0.006, 0.010] # 0%, 0.3%, 0.6%, 1%
for i, spacing in enumerate(spacings):
levels.append({
'price': round(entry_price * (1 - spacing), 2),
'ratio': [0.4, 0.3, 0.2, 0.1][i],
'reasoning': f'Level {i+1}' if i > 0 else 'Initial entry',
'level': i,
})
else: # SHORT
# For SHORT: first entry lowest, subsequent entries higher
spacings = [0, 0.003, 0.006, 0.010]
for i, spacing in enumerate(spacings):
levels.append({
'price': round(entry_price * (1 + spacing), 2),
'ratio': [0.4, 0.3, 0.2, 0.1][i],
'reasoning': f'Level {i+1}' if i > 0 else 'Initial entry',
'level': i,
})
if direction == 'LONG':
price = round(entry_price * (1 - spacing), 2)
else:
price = round(entry_price * (1 + spacing), 2)
levels.append({'price': price, 'ratio': [0.4, 0.3, 0.2, 0.1][i], 'level': i})
return levels
# Try new format first
# Parse opportunities
opportunities = llm_decision.get('opportunities', {})
short_term = opportunities.get('short_term_5m_15m_1h', {})
medium_term = opportunities.get('medium_term_4h_1d', {})
long_term = opportunities.get('long_term_1d_1w', {})
ambush = opportunities.get('ambush', {})
# Fallback to old format for backward compatibility
if not short_term and not medium_term and not long_term:
intraday = opportunities.get('intraday', {})
swing = opportunities.get('swing', {})
# Map old format to new format
short_term = intraday
medium_term = swing
long_term = {}
short_term = opportunities.get('intraday', {})
medium_term = opportunities.get('swing', {})
# Apply minimum profit filter to all opportunities - 不同周期不同要求
MIN_PROFIT_SHORT = settings.MIN_PROFIT_PCT_SHORT # 短周期 1%
MIN_PROFIT_MEDIUM = settings.MIN_PROFIT_PCT_MEDIUM # 中周期 2%
MIN_PROFIT_LONG = settings.MIN_PROFIT_PCT_LONG # 长周期 5%
# Apply profit filters
MIN_PROFIT_SHORT = settings.MIN_PROFIT_PCT_SHORT
MIN_PROFIT_MEDIUM = settings.MIN_PROFIT_PCT_MEDIUM
MIN_PROFIT_LONG = settings.MIN_PROFIT_PCT_LONG
# Filter short_term (最低 1%)
short_term_valid = meets_profit_threshold(short_term, MIN_PROFIT_SHORT)
if short_term.get('exists') and not short_term_valid:
profit_pct = calc_profit_pct(
safe_float(short_term.get('entry_price'), 0),
safe_float(short_term.get('take_profit'), 0),
short_term.get('direction')
)
logger.info(f"短期机会被过滤: 盈利空间 {profit_pct:.2f}% < {MIN_PROFIT_SHORT}%")
short_term = {'exists': False, 'reasoning': f'盈利空间不足{MIN_PROFIT_SHORT}% (仅{profit_pct:.2f}%)，建议观望'}
if short_term.get('exists') and not meets_profit_threshold(short_term, MIN_PROFIT_SHORT):
short_term = {'exists': False, 'reasoning': '盈利空间不足'}
if medium_term.get('exists') and not meets_profit_threshold(medium_term, MIN_PROFIT_MEDIUM):
medium_term = {'exists': False, 'reasoning': '盈利空间不足'}
if long_term.get('exists') and not meets_profit_threshold(long_term, MIN_PROFIT_LONG):
long_term = {'exists': False, 'reasoning': '盈利空间不足'}
# Filter medium_term (最低 2%)
medium_term_valid = meets_profit_threshold(medium_term, MIN_PROFIT_MEDIUM)
if medium_term.get('exists') and not medium_term_valid:
profit_pct = calc_profit_pct(
safe_float(medium_term.get('entry_price'), 0),
safe_float(medium_term.get('take_profit'), 0),
medium_term.get('direction')
)
logger.info(f"中期机会被过滤: 盈利空间 {profit_pct:.2f}% < {MIN_PROFIT_MEDIUM}%")
medium_term = {'exists': False, 'reasoning': f'盈利空间不足{MIN_PROFIT_MEDIUM}% (仅{profit_pct:.2f}%)，建议观望'}
# Normalize entry levels
short_levels = normalize_entry_levels(short_term, short_term.get('direction', 'LONG')) if short_term.get('exists') else []
medium_levels = normalize_entry_levels(medium_term, medium_term.get('direction', 'LONG')) if medium_term.get('exists') else []
long_levels = normalize_entry_levels(long_term, long_term.get('direction', 'LONG')) if long_term.get('exists') else []
# Filter long_term (最低 5%)
long_term_valid = meets_profit_threshold(long_term, MIN_PROFIT_LONG)
if long_term.get('exists') and not long_term_valid:
profit_pct = calc_profit_pct(
safe_float(long_term.get('entry_price'), 0),
safe_float(long_term.get('take_profit'), 0),
long_term.get('direction')
)
logger.info(f"长期机会被过滤: 盈利空间 {profit_pct:.2f}% < {MIN_PROFIT_LONG}%")
long_term = {'exists': False, 'reasoning': f'盈利空间不足{MIN_PROFIT_LONG}% (仅{profit_pct:.2f}%)，建议观望'}
def get_first_entry(levels, fallback):
return levels[0]['price'] if levels else fallback
# Determine primary levels (priority: short > medium > long) - 优先日内短线
entry = market_context.get('current_price', 0)
stop_loss = 0
take_profit = 0
if short_term.get('exists'):
entry = safe_float(short_term.get('entry_price'), market_context.get('current_price', 0))
stop_loss = safe_float(short_term.get('stop_loss'), 0)
take_profit = safe_float(short_term.get('take_profit'), 0)
elif medium_term.get('exists'):
entry = safe_float(medium_term.get('entry_price'), market_context.get('current_price', 0))
stop_loss = safe_float(medium_term.get('stop_loss'), 0)
take_profit = safe_float(medium_term.get('take_profit'), 0)
elif long_term.get('exists'):
entry = safe_float(long_term.get('entry_price'), market_context.get('current_price', 0))
stop_loss = safe_float(long_term.get('stop_loss'), 0)
take_profit = safe_float(long_term.get('take_profit'), 0)
# Get recommendations by timeframe
# Build decision
recommendations = llm_decision.get('recommendations_by_timeframe', {})
# Get current price for entry level normalization
current_price = market_context.get('current_price', 0)
# Normalize entry_levels for each opportunity
short_term_levels = normalize_entry_levels(
short_term, short_term.get('direction', 'LONG'), current_price
) if short_term.get('exists') else []
medium_term_levels = normalize_entry_levels(
medium_term, medium_term.get('direction', 'LONG'), current_price
) if medium_term.get('exists') else []
long_term_levels = normalize_entry_levels(
long_term, long_term.get('direction', 'LONG'), current_price
) if long_term.get('exists') else []
# Get first entry price for backward compatibility
def get_first_entry(levels: list, fallback: float) -> float:
if levels and len(levels) > 0:
return levels[0].get('price', fallback)
return fallback
# Validate and structure decision
decision = {
'timestamp': datetime.now().isoformat(),
'signal_type': llm_decision.get('signal', 'HOLD').upper(),
'confidence': safe_float(llm_decision.get('confidence'), 0.5),
'trade_type': 'MULTI_TIMEFRAME', # New format uses multiple timeframes
'trade_type': 'MULTI_TIMEFRAME',
'reasoning': llm_decision.get('reasoning', ''),
# New opportunities breakdown (multi-timeframe) with entry_levels
'opportunities': {
'short_term_5m_15m_1h': {
'exists': short_term.get('exists', False),
'direction': short_term.get('direction'),
'entry_levels': short_term_levels, # New: array of entry levels for pyramiding
'entry_price': get_first_entry(short_term_levels, safe_float(short_term.get('entry_price'), 0)), # Backward compat
'entry_levels': short_levels,
'entry_price': get_first_entry(short_levels, safe_float(short_term.get('entry_price'), 0)),
'stop_loss': safe_float(short_term.get('stop_loss'), 0),
'take_profit': safe_float(short_term.get('take_profit'), 0),
'reasoning': short_term.get('reasoning', '')
@ -622,8 +709,8 @@ ${current_price:,.2f}
'medium_term_4h_1d': {
'exists': medium_term.get('exists', False),
'direction': medium_term.get('direction'),
'entry_levels': medium_term_levels,
'entry_price': get_first_entry(medium_term_levels, safe_float(medium_term.get('entry_price'), 0)),
'entry_levels': medium_levels,
'entry_price': get_first_entry(medium_levels, safe_float(medium_term.get('entry_price'), 0)),
'stop_loss': safe_float(medium_term.get('stop_loss'), 0),
'take_profit': safe_float(medium_term.get('take_profit'), 0),
'reasoning': medium_term.get('reasoning', '')
@ -631,23 +718,17 @@ ${current_price:,.2f}
'long_term_1d_1w': {
'exists': long_term.get('exists', False),
'direction': long_term.get('direction'),
'entry_levels': long_term_levels,
'entry_price': get_first_entry(long_term_levels, safe_float(long_term.get('entry_price'), 0)),
'entry_levels': long_levels,
'entry_price': get_first_entry(long_levels, safe_float(long_term.get('entry_price'), 0)),
'stop_loss': safe_float(long_term.get('stop_loss'), 0),
'take_profit': safe_float(long_term.get('take_profit'), 0),
'reasoning': long_term.get('reasoning', '')
},
'ambush': {
'exists': ambush.get('exists', False),
'price_level': safe_float(ambush.get('price_level'), 0),
'reasoning': ambush.get('reasoning', '')
},
# Keep old format for backward compatibility
'intraday': {
'exists': short_term.get('exists', False),
'direction': short_term.get('direction'),
'entry_levels': short_term_levels,
'entry_price': get_first_entry(short_term_levels, safe_float(short_term.get('entry_price'), 0)),
'entry_levels': short_levels,
'entry_price': get_first_entry(short_levels, safe_float(short_term.get('entry_price'), 0)),
'stop_loss': safe_float(short_term.get('stop_loss'), 0),
'take_profit': safe_float(short_term.get('take_profit'), 0),
'reasoning': short_term.get('reasoning', '')
@ -655,7 +736,7 @@ ${current_price:,.2f}
'swing': {
'exists': medium_term.get('exists', False) or long_term.get('exists', False),
'direction': medium_term.get('direction') or long_term.get('direction'),
'entry_levels': medium_term_levels if medium_term.get('exists') else long_term_levels,
'entry_levels': medium_levels if medium_term.get('exists') else long_levels,
'entry_price': safe_float(medium_term.get('entry_price') or long_term.get('entry_price'), 0),
'stop_loss': safe_float(medium_term.get('stop_loss') or long_term.get('stop_loss'), 0),
'take_profit': safe_float(medium_term.get('take_profit') or long_term.get('take_profit'), 0),
@ -663,7 +744,6 @@ ${current_price:,.2f}
},
},
# Recommendations by timeframe
'recommendations_by_timeframe': {
'short_term': recommendations.get('short_term', ''),
'medium_term': recommendations.get('medium_term', ''),
@ -672,16 +752,16 @@ ${current_price:,.2f}
# Primary levels (for backward compatibility)
'levels': {
'current_price': market_context.get('current_price', 0),
'entry': entry,
'stop_loss': stop_loss,
'take_profit_1': take_profit,
'take_profit_2': take_profit,
'take_profit_3': take_profit,
'current_price': current_price,
'entry': get_first_entry(short_levels, 0) or get_first_entry(medium_levels, 0) or get_first_entry(long_levels, 0),
'stop_loss': safe_float(short_term.get('stop_loss'), 0) or safe_float(medium_term.get('stop_loss'), 0) or safe_float(long_term.get('stop_loss'), 0),
'take_profit_1': safe_float(short_term.get('take_profit'), 0) or safe_float(medium_term.get('take_profit'), 0) or safe_float(long_term.get('take_profit'), 0),
'take_profit_2': 0,
'take_profit_3': 0,
},
'risk_level': llm_decision.get('risk_level', 'MEDIUM'),
'key_factors': llm_decision.get('key_factors', []),
'raw_response': response_text,
'raw_response': '',
}
# Calculate risk-reward ratio

164
trading/paper_trading.py
View File

@ -798,7 +798,12 @@ class MultiTimeframePaperTrader:
def _check_higher_timeframe_trend(
self, symbol: str, tf: TimeFrame, direction: str, signal: Dict
) -> Dict:
"""检查大周期趋势是否与当前方向一致"""
"""检查大周期趋势是否与当前方向一致
支持两种格式:
1. 新格式: trades数组
2. 旧格式: opportunities对象
"""
higher_tfs = TIMEFRAME_HIERARCHY.get(tf, [])
if not higher_tfs:
@ -806,6 +811,33 @@ class MultiTimeframePaperTrader:
# 从信号中获取各周期的方向
llm_signal = signal.get('llm_signal') or signal.get('aggregated_signal', {}).get('llm_signal', {})
if not llm_signal:
return {'aligned': True, 'reason': '无LLM信号数据'}
# ========== 新格式: trades数组 ==========
trades = llm_signal.get('trades', [])
if trades and isinstance(trades, list) and len(trades) >= 3:
trades_by_tf = {t.get('timeframe'): t for t in trades if t.get('timeframe')}
for higher_tf in higher_tfs:
higher_tf_key = higher_tf.value # 'short', 'medium', 'long'
higher_trade = trades_by_tf.get(higher_tf_key, {})
if higher_trade and higher_trade.get('status') == 'ACTIVE':
higher_direction = higher_trade.get('direction')
if higher_direction and higher_direction != direction and higher_direction != 'NONE':
return {
'aligned': False,
'higher_tf': higher_tf.value,
'higher_tf_trend': higher_direction,
'current_direction': direction,
'reason': f'{higher_tf.value}周期为{higher_direction}，与{direction}冲突',
}
return {'aligned': True, 'reason': '大周期趋势一致或无明确方向'}
# ========== 旧格式: opportunities对象 ==========
opportunities = llm_signal.get('opportunities', {}) if llm_signal else {}
for higher_tf in higher_tfs:
@ -989,12 +1021,43 @@ class MultiTimeframePaperTrader:
def _extract_timeframe_signal(
self, signal: Dict[str, Any], signal_keys: List[str]
) -> Optional[Dict[str, Any]]:
"""提取特定周期的信号"""
"""提取特定周期的信号
支持两种格式:
1. 新格式: trades数组 (优先)
2. 旧格式: opportunities对象 (向后兼容)
"""
try:
# 从 llm_signal.opportunities 中提取
# 从 llm_signal 中提取
llm_signal = signal.get('llm_signal') or signal.get('aggregated_signal', {}).get('llm_signal')
if llm_signal and isinstance(llm_signal, dict):
# ========== 新格式: trades数组 ==========
trades = llm_signal.get('trades', [])
if trades and isinstance(trades, list) and len(trades) >= 3:
# 确定当前 signal_keys 对应的 timeframe
tf_mapping = {
'short_term_5m_15m_1h': 'short',
'intraday': 'short',
'medium_term_4h_1d': 'medium',
'swing': 'medium',
'long_term_1d_1w': 'long',
}
target_tf = None
for key in signal_keys:
if key in tf_mapping:
target_tf = tf_mapping[key]
break
if target_tf:
# 从 trades 数组中找到对应周期
for trade in trades:
if trade.get('timeframe') == target_tf:
# 转换为统一格式
return self._convert_trade_to_opportunity(trade)
# ========== 旧格式: opportunities对象 ==========
opportunities = llm_signal.get('opportunities', {})
for key in signal_keys:
if key in opportunities and opportunities[key]:
@ -1005,6 +1068,27 @@ class MultiTimeframePaperTrader:
if agg:
llm = agg.get('llm_signal', {})
if llm:
# 先检查新格式
trades = llm.get('trades', [])
if trades and isinstance(trades, list) and len(trades) >= 3:
tf_mapping = {
'short_term_5m_15m_1h': 'short',
'intraday': 'short',
'medium_term_4h_1d': 'medium',
'swing': 'medium',
'long_term_1d_1w': 'long',
}
target_tf = None
for key in signal_keys:
if key in tf_mapping:
target_tf = tf_mapping[key]
break
if target_tf:
for trade in trades:
if trade.get('timeframe') == target_tf:
return self._convert_trade_to_opportunity(trade)
# 回退到旧格式
opps = llm.get('opportunities', {})
for key in signal_keys:
if key in opps and opps[key]:
@ -1015,6 +1099,80 @@ class MultiTimeframePaperTrader:
logger.error(f"Error extracting signal: {e}")
return None
def _convert_trade_to_opportunity(self, trade: Dict[str, Any]) -> Dict[str, Any]:
"""将新格式 trade 转换为旧格式 opportunity
新格式:
{
"id": "short_001",
"timeframe": "short",
"status": "ACTIVE|INACTIVE",
"direction": "LONG|SHORT|NONE",
"entry": {"price_1": 90000, "price_2": 89700, ...},
"exit": {"stop_loss": 88500, "take_profit_1": 91000, ...},
"position": {"size_pct_1": 40, "size_pct_2": 30, ...},
"risk_reward": 2.5,
"expected_profit_pct": 1.5,
"reasoning": "..."
}
转换为:
{
"exists": True,
"direction": "LONG",
"entry_price": 90000,
"entry_levels": [...],
"stop_loss": 88500,
"take_profit": 91000,
"reasoning": "..."
}
"""
status = trade.get('status', 'INACTIVE')
is_active = status == 'ACTIVE'
if not is_active:
return {
'exists': False,
'direction': None,
'entry_price': 0,
'stop_loss': 0,
'take_profit': 0,
'reasoning': trade.get('reasoning', ''),
}
entry = trade.get('entry', {})
exit_data = trade.get('exit', {})
position = trade.get('position', {})
# 构建 entry_levels（金字塔入场价位）
entry_levels = []
for i in range(1, 5):
price = entry.get(f'price_{i}', 0)
ratio = position.get(f'size_pct_{i}', [40, 30, 20, 10][i-1]) / 100
if price > 0:
entry_levels.append({
'price': float(price),
'ratio': ratio,
'level': i - 1,
})
# 第一个入场价作为主入场价
entry_price = float(entry.get('price_1', 0))
return {
'exists': True,
'direction': trade.get('direction', 'NONE'),
'entry_price': entry_price,
'entry_levels': entry_levels,
'stop_loss': float(exit_data.get('stop_loss', 0)),
'take_profit': float(exit_data.get('take_profit_1', 0)),
'take_profit_2': float(exit_data.get('take_profit_2', 0)),
'take_profit_3': float(exit_data.get('take_profit_3', 0)),
'risk_reward': trade.get('risk_reward', 0),
'expected_profit_pct': trade.get('expected_profit_pct', 0),
'reasoning': trade.get('reasoning', ''),
}
def _get_max_position_value(self, symbol: str, tf: TimeFrame) -> float:
"""获取最大仓位价值（本金 × 杠杆）"""
account = self.accounts[symbol][tf]