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

1

Browse Source
This commit is contained in:
aaron 2025-12-11 16:11:35 +08:00
parent 672fea3d8e
commit e21be1b946
3 changed files with 565 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
docker-compose.yml
View File

@ -12,12 +12,18 @@ services:
volumes:
- ./output:/app/output # 输出信号文件
environment:
# Symbol Configuration
- SYMBOL=BTCUSDT
# Symbol Configuration - 多币种支持
- SYMBOLS=BTCUSDT,ETHUSDT
- SYMBOL=BTCUSDT # 向后兼容
# Signal generation interval
- SIGNAL_INTERVAL_MINUTES=15 # 每15分钟生成一次信号
# Volatility trigger - 波动率触发
- ENABLE_VOLATILITY_TRIGGER=true
- VOLATILITY_THRESHOLD=0.5
- VOLATILITY_COOLDOWN_MINUTES=3
# Note: LLM API and DingTalk configs are loaded from .env file
- LOG_LEVEL=INFO
@ -42,7 +48,9 @@ services:
volumes:
- ./output:/app/output # 共享信号文件和交易状态
environment:
- SYMBOL=BTCUSDT
# Symbol Configuration - 多币种支持
- SYMBOLS=BTCUSDT,ETHUSDT
- SYMBOL=BTCUSDT # 向后兼容
- LOG_LEVEL=INFO
depends_on:
- scheduler
@ -68,6 +76,9 @@ services:
ports:
- "18080:8000" # 使用18080端口避免冲突
environment:
# Symbol Configuration - 多币种支持
- SYMBOLS=BTCUSDT,ETHUSDT
- SYMBOL=BTCUSDT # 向后兼容
- LOG_LEVEL=INFO
depends_on:
- paper-trading

117
signals/llm_decision.py
View File

@ -181,12 +181,17 @@ ${current_price:,.2f}
}}
}},
// 分时间级别的交易机会分析
// 分时间级别的交易机会分析 - 支持金字塔加仓的多级进场
"opportunities": {{
"short_term_5m_15m_1h": {{
"exists": true/false,
"direction": "LONG" | "SHORT" | null,
"entry_price": 进场价格数值或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": "短期日内机会说明"
@ -194,7 +199,12 @@ ${current_price:,.2f}
"medium_term_4h_1d": {{
"exists": true/false,
"direction": "LONG" | "SHORT" | null,
"entry_price": 进场价格数值或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": "中期波段机会说明"
@ -202,7 +212,12 @@ ${current_price:,.2f}
"long_term_1d_1w": {{
"exists": true/false,
"direction": "LONG" | "SHORT" | null,
"entry_price": 进场价格数值或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": "长期趋势机会说明"
@ -231,6 +246,11 @@ ${current_price:,.2f}
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%
"""
@ -432,6 +452,60 @@ ${current_price:,.2f}
# 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"""
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
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
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,
})
return levels
# Try new format first
short_term = opportunities.get('short_term_5m_15m_1h', {})
medium_term = opportunities.get('medium_term_4h_1d', {})
@ -506,6 +580,26 @@ ${current_price:,.2f}
# Get recommendations by timeframe
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(),
@ -514,12 +608,13 @@ ${current_price:,.2f}
'trade_type': 'MULTI_TIMEFRAME', # New format uses multiple timeframes
'reasoning': llm_decision.get('reasoning', ''),
# New opportunities breakdown (multi-timeframe)
# 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_price': safe_float(short_term.get('entry_price'), 0),
'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
'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', '')
@ -527,7 +622,8 @@ ${current_price:,.2f}
'medium_term_4h_1d': {
'exists': medium_term.get('exists', False),
'direction': medium_term.get('direction'),
'entry_price': safe_float(medium_term.get('entry_price'), 0),
'entry_levels': medium_term_levels,
'entry_price': get_first_entry(medium_term_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', '')
@ -535,7 +631,8 @@ ${current_price:,.2f}
'long_term_1d_1w': {
'exists': long_term.get('exists', False),
'direction': long_term.get('direction'),
'entry_price': safe_float(long_term.get('entry_price'), 0),
'entry_levels': long_term_levels,
'entry_price': get_first_entry(long_term_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', '')
@ -549,7 +646,8 @@ ${current_price:,.2f}
'intraday': {
'exists': short_term.get('exists', False),
'direction': short_term.get('direction'),
'entry_price': safe_float(short_term.get('entry_price'), 0),
'entry_levels': short_term_levels,
'entry_price': get_first_entry(short_term_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', '')
@ -557,6 +655,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_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),

457
trading/paper_trading.py
View File

@ -38,30 +38,52 @@ TIMEFRAME_CONFIG = {
'name_en': 'Short-term',
'signal_keys': ['short_term_5m_15m_1h', 'intraday'],
'leverage': 10,
'initial_balance': 10000.0, # 独立初始资金
'max_price_deviation': 0.001, # 0.1% - 短周期要求精准入场
'initial_balance': 10000.0,
'signal_expiry_minutes': 5, # 信号有效期5分钟
'min_risk_reward_ratio': 1.5, # 最小风险回报比
'base_price_deviation': 0.003, # 基础价格偏差 0.3%
'atr_deviation_multiplier': 0.5, # ATR偏差系数
},
TimeFrame.MEDIUM: {
'name': '中周期',
'name_en': 'Medium-term',
'signal_keys': ['medium_term_4h_1d', 'swing'],
'leverage': 10,
'initial_balance': 10000.0, # 独立初始资金
'max_price_deviation': 0.003, # 0.3% - 中周期适中容错
'initial_balance': 10000.0,
'signal_expiry_minutes': 30, # 信号有效期30分钟
'min_risk_reward_ratio': 1.5,
'base_price_deviation': 0.005, # 基础价格偏差 0.5%
'atr_deviation_multiplier': 0.8,
},
TimeFrame.LONG: {
'name': '长周期',
'name_en': 'Long-term',
'signal_keys': ['long_term_1d_1w'],
'leverage': 10,
'initial_balance': 10000.0, # 独立初始资金
'max_price_deviation': 0.005, # 0.5% - 长周期追求大趋势
'initial_balance': 10000.0,
'signal_expiry_minutes': 120, # 信号有效期2小时
'min_risk_reward_ratio': 2.0, # 长周期要求更高回报比
'base_price_deviation': 0.01, # 基础价格偏差 1%
'atr_deviation_multiplier': 1.0,
},
}
# 金字塔加仓配置：每次加仓的仓位比例（总计100%）
PYRAMID_LEVELS = [0.4, 0.3, 0.2, 0.1] # 首仓40%，加仓30%、20%、10%
# 加仓价格改善要求（相对于均价的百分比）
PYRAMID_PRICE_IMPROVEMENT = 0.005 # 加仓价格需比均价优 0.5%
# 多周期协调配置
TIMEFRAME_HIERARCHY = {
TimeFrame.SHORT: [TimeFrame.MEDIUM, TimeFrame.LONG], # 短周期受中、长周期约束
TimeFrame.MEDIUM: [TimeFrame.LONG], # 中周期受长周期约束
TimeFrame.LONG: [], # 长周期不受约束
}
# 信号确认配置
SIGNAL_CONFIRMATION_COUNT = 2 # 需要连续2次相同方向信号才执行
@dataclass
class PositionEntry:
@ -177,6 +199,23 @@ class Trade:
return cls(**data)
@dataclass
class SignalHistory:
"""信号历史记录（用于信号确认机制）"""
direction: str # LONG, SHORT, NONE
timestamp: str
entry_price: float = 0.0
stop_loss: float = 0.0
take_profit: float = 0.0
def to_dict(self) -> dict:
return asdict(self)
@classmethod
def from_dict(cls, data: dict) -> 'SignalHistory':
return cls(**data)
@dataclass
class TimeFrameAccount:
"""单个币种单个周期的账户
@ -199,6 +238,8 @@ class TimeFrameAccount:
trades: List[Trade] = field(default_factory=list)
stats: Dict = field(default_factory=dict)
equity_curve: List[Dict] = field(default_factory=list)
signal_history: List[SignalHistory] = field(default_factory=list) # 信号确认历史
last_atr: float = 0.0 # 最近的ATR值（用于动态计算）
def __post_init__(self):
if not self.stats:
@ -453,7 +494,7 @@ class MultiTimeframePaperTrader:
def _process_timeframe_signal(
self, symbol: str, tf: TimeFrame, signal: Dict[str, Any], current_price: float
) -> Dict[str, Any]:
"""处理单个币种单个周期的信号"""
"""处理单个币种单个周期的信号（包含所有优化）"""
account = self.accounts[symbol][tf]
config = TIMEFRAME_CONFIG[tf]
@ -477,43 +518,99 @@ class MultiTimeframePaperTrader:
tf_signal = self._extract_timeframe_signal(signal, config['signal_keys'])
if not tf_signal or not tf_signal.get('exists'):
# 记录无信号到历史
self._record_signal_history(account, 'NONE', 0, 0, 0)
result['action'] = 'NO_SIGNAL'
return result
direction = tf_signal.get('direction')
if not direction:
self._record_signal_history(account, 'NONE', 0, 0, 0)
result['action'] = 'NO_SIGNAL'
return result
signal_stop_loss = tf_signal.get('stop_loss', 0)
signal_take_profit = tf_signal.get('take_profit', 0)
signal_entry_price = tf_signal.get('entry_price', 0)
signal_timestamp = signal.get('timestamp') or signal.get('aggregated_signal', {}).get('timestamp')
# 验证止盈止损
# ========== 优化1: 信号时效性检查 ==========
if signal_timestamp:
expiry_check = self._check_signal_expiry(signal_timestamp, config)
if not expiry_check['valid']:
result['action'] = 'SIGNAL_EXPIRED'
result['details'] = expiry_check
logger.info(f"[{symbol}][{config['name']}] 信号已过期: {expiry_check['age_minutes']:.1f}分钟")
return result
# 验证止盈止损存在
if signal_stop_loss <= 0 or signal_take_profit <= 0:
result['action'] = 'NO_SIGNAL'
result['details'] = {'reason': '缺少有效止盈止损'}
return result
# 检查价格偏差：当前价格与建议入场价偏差超过阈值则不开仓
max_deviation = config.get('max_price_deviation', 0.002)
# ========== 优化2: 风险回报比验证 ==========
rr_check = self._check_risk_reward_ratio(
direction, current_price, signal_stop_loss, signal_take_profit, config
)
if not rr_check['valid']:
result['action'] = 'LOW_RISK_REWARD'
result['details'] = rr_check
logger.info(
f"[{symbol}][{config['name']}] 风险回报比不足: {rr_check['ratio']:.2f} < {rr_check['min_ratio']}"
)
return result
# ========== 优化3: 动态价格偏差（基于ATR） ==========
# 更新ATR（从信号中获取）
atr = self._get_atr_from_signal(signal)
if atr > 0:
account.last_atr = atr
max_deviation = self._calculate_dynamic_deviation(config, account.last_atr, current_price)
if signal_entry_price > 0:
price_deviation = abs(current_price - signal_entry_price) / signal_entry_price
if price_deviation > max_deviation:
result['action'] = 'PRICE_DEVIATION'
result['details'] = {
'reason': f'价格偏差过大: {price_deviation*100:.2f}% > {max_deviation*100:.1f}%',
'reason': f'价格偏差过大: {price_deviation*100:.2f}% > {max_deviation*100:.2f}%',
'signal_entry': signal_entry_price,
'current_price': current_price,
'deviation_pct': price_deviation * 100,
'max_deviation_pct': max_deviation * 100,
'atr_used': account.last_atr,
}
logger.info(
f"[{config['name']}] 跳过开仓: 价格偏差 {price_deviation*100:.2f}% > {max_deviation*100:.1f}% "
f"[{symbol}][{config['name']}] 跳过开仓: 价格偏差 {price_deviation*100:.2f}% > {max_deviation*100:.2f}% "
f"(信号价: ${signal_entry_price:.2f}, 当前价: ${current_price:.2f})"
)
return result
# ========== 优化4: 多周期协调（大周期趋势过滤） ==========
if not account.position or account.position.side == 'FLAT':
trend_check = self._check_higher_timeframe_trend(symbol, tf, direction, signal)
if not trend_check['aligned']:
result['action'] = 'TREND_CONFLICT'
result['details'] = trend_check
logger.info(
f"[{symbol}][{config['name']}] 与大周期趋势冲突: {direction} vs {trend_check['higher_tf_trend']}"
)
return result
# ========== 优化5: 信号确认机制 ==========
self._record_signal_history(account, direction, signal_entry_price, signal_stop_loss, signal_take_profit)
if not account.position or account.position.side == 'FLAT':
confirm_check = self._check_signal_confirmation(account, direction)
if not confirm_check['confirmed']:
result['action'] = 'AWAITING_CONFIRMATION'
result['details'] = confirm_check
logger.debug(
f"[{symbol}][{config['name']}] 等待信号确认: {confirm_check['count']}/{SIGNAL_CONFIRMATION_COUNT}"
)
return result
# 3. 如果有持仓
if account.position and account.position.side != 'FLAT':
# 反向信号：只平仓不开反向仓
@ -527,13 +624,19 @@ class MultiTimeframePaperTrader:
)
return result
else:
# 同方向信号：尝试金字塔加仓
add_result = self._add_position(
# ========== 优化6: 加仓价格检查（支持entry_levels）==========
entry_levels = tf_signal.get('entry_levels', [])
add_result = self._add_position_with_price_check(
symbol, tf, current_price,
signal_stop_loss, signal_take_profit,
tf_signal.get('reasoning', '')[:100]
tf_signal.get('reasoning', '')[:100],
entry_levels=entry_levels
)
if add_result:
if add_result.get('skipped'):
result['action'] = 'ADD_PRICE_NOT_IMPROVED'
result['details'] = add_result
else:
result['action'] = 'ADD'
result['details'] = add_result
else:
@ -547,9 +650,55 @@ class MultiTimeframePaperTrader:
return result
# 4. 无持仓，开新仓（首仓）
# ========== 优化7: 动态止损（验证止损距离合理性） ==========
adjusted_sl, adjusted_tp = self._adjust_stop_loss_take_profit(
direction, current_price, signal_stop_loss, signal_take_profit,
account.last_atr, config
)
# ========== 优化8: 检查首仓价格是否匹配 entry_levels ==========
entry_levels = tf_signal.get('entry_levels', [])
if entry_levels and len(entry_levels) > 0:
first_entry = entry_levels[0]
target_price = first_entry.get('price', 0)
if target_price > 0:
price_tolerance = 0.003 # 0.3% 容差
if direction == 'LONG':
# 做多：当前价格需要 ≤ 首仓目标价格
if current_price > target_price * (1 + price_tolerance):
result['action'] = 'WAIT_ENTRY_LEVEL'
result['details'] = {
'reason': f'等待首仓价位 ${target_price:.2f}',
'current_price': current_price,
'target_price': target_price,
'direction': direction,
'entry_levels': entry_levels,
}
logger.info(
f"[{symbol}][{config['name']}] 等待首仓价位: "
f"目标=${target_price:.2f}, 当前=${current_price:.2f}"
)
return result
else: # SHORT
# 做空：当前价格需要 ≥ 首仓目标价格
if current_price < target_price * (1 - price_tolerance):
result['action'] = 'WAIT_ENTRY_LEVEL'
result['details'] = {
'reason': f'等待首仓价位 ${target_price:.2f}',
'current_price': current_price,
'target_price': target_price,
'direction': direction,
'entry_levels': entry_levels,
}
logger.info(
f"[{symbol}][{config['name']}] 等待首仓价位: "
f"目标=${target_price:.2f}, 当前=${current_price:.2f}"
)
return result
open_result = self._open_position(
symbol, tf, direction, current_price,
signal_stop_loss, signal_take_profit,
adjusted_sl, adjusted_tp,
tf_signal.get('reasoning', '')[:100]
)
@ -561,6 +710,282 @@ class MultiTimeframePaperTrader:
return result
# ==================== 新增优化方法 ====================
def _check_signal_expiry(self, signal_timestamp: str, config: Dict) -> Dict:
"""检查信号是否过期"""
try:
# 解析信号时间
if 'T' in signal_timestamp:
signal_time = datetime.fromisoformat(signal_timestamp.replace('Z', '+00:00'))
else:
signal_time = datetime.fromisoformat(signal_timestamp)
# 移除时区信息进行比较
if signal_time.tzinfo:
signal_time = signal_time.replace(tzinfo=None)
now = datetime.now()
age = now - signal_time
age_minutes = age.total_seconds() / 60
expiry_minutes = config.get('signal_expiry_minutes', 15)
return {
'valid': age_minutes <= expiry_minutes,
'age_minutes': age_minutes,
'expiry_minutes': expiry_minutes,
'signal_time': signal_timestamp,
}
except Exception as e:
logger.warning(f"信号时间解析失败: {e}")
return {'valid': True, 'age_minutes': 0, 'expiry_minutes': 0}
def _check_risk_reward_ratio(
self, direction: str, entry_price: float,
stop_loss: float, take_profit: float, config: Dict
) -> Dict:
"""验证风险回报比"""
min_ratio = config.get('min_risk_reward_ratio', 1.5)
if direction == 'LONG':
risk = entry_price - stop_loss
reward = take_profit - entry_price
else: # SHORT
risk = stop_loss - entry_price
reward = entry_price - take_profit
if risk <= 0:
return {'valid': False, 'ratio': 0, 'min_ratio': min_ratio, 'reason': '止损设置错误'}
ratio = reward / risk
return {
'valid': ratio >= min_ratio,
'ratio': round(ratio, 2),
'min_ratio': min_ratio,
'risk': risk,
'reward': reward,
}
def _calculate_dynamic_deviation(self, config: Dict, atr: float, current_price: float) -> float:
"""计算动态价格偏差阈值"""
base_deviation = config.get('base_price_deviation', 0.005)
atr_multiplier = config.get('atr_deviation_multiplier', 0.5)
if atr > 0 and current_price > 0:
# ATR 百分比
atr_pct = atr / current_price
# 动态偏差 = 基础偏差 + ATR偏差
dynamic_deviation = base_deviation + (atr_pct * atr_multiplier)
return min(dynamic_deviation, 0.02) # 最大2%
else:
return base_deviation
def _get_atr_from_signal(self, signal: Dict) -> float:
"""从信号中提取ATR值"""
try:
# 尝试多个路径
atr = signal.get('market_analysis', {}).get('volatility_analysis', {}).get('atr', 0)
if not atr:
atr = signal.get('aggregated_signal', {}).get('levels', {}).get('atr', 0)
if not atr:
atr = signal.get('quantitative_signal', {}).get('indicators', {}).get('atr', 0)
return float(atr) if atr else 0.0
except:
return 0.0
def _check_higher_timeframe_trend(
self, symbol: str, tf: TimeFrame, direction: str, signal: Dict
) -> Dict:
"""检查大周期趋势是否与当前方向一致"""
higher_tfs = TIMEFRAME_HIERARCHY.get(tf, [])
if not higher_tfs:
return {'aligned': True, 'reason': '无需检查大周期'}
# 从信号中获取各周期的方向
llm_signal = signal.get('llm_signal') or signal.get('aggregated_signal', {}).get('llm_signal', {})
opportunities = llm_signal.get('opportunities', {}) if llm_signal else {}
for higher_tf in higher_tfs:
higher_config = TIMEFRAME_CONFIG[higher_tf]
for key in higher_config['signal_keys']:
higher_opp = opportunities.get(key, {})
if higher_opp and higher_opp.get('exists'):
higher_direction = higher_opp.get('direction')
if higher_direction and higher_direction != direction:
# 大周期方向相反，不建议开仓
# 但如果大周期是 HOLD/观望，则允许
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': '大周期趋势一致或无明确方向'}
def _record_signal_history(
self, account: TimeFrameAccount, direction: str,
entry_price: float, stop_loss: float, take_profit: float
):
"""记录信号历史"""
history = SignalHistory(
direction=direction,
timestamp=datetime.now().isoformat(),
entry_price=entry_price,
stop_loss=stop_loss,
take_profit=take_profit,
)
account.signal_history.append(history)
# 只保留最近10条
if len(account.signal_history) > 10:
account.signal_history = account.signal_history[-10:]
def _check_signal_confirmation(self, account: TimeFrameAccount, direction: str) -> Dict:
"""检查信号是否已确认（连续N次相同方向）"""
if len(account.signal_history) < SIGNAL_CONFIRMATION_COUNT:
return {
'confirmed': False,
'count': len(account.signal_history),
'required': SIGNAL_CONFIRMATION_COUNT,
}
# 检查最近N次信号是否都是同一方向
recent = account.signal_history[-SIGNAL_CONFIRMATION_COUNT:]
same_direction_count = sum(1 for h in recent if h.direction == direction)
return {
'confirmed': same_direction_count >= SIGNAL_CONFIRMATION_COUNT,
'count': same_direction_count,
'required': SIGNAL_CONFIRMATION_COUNT,
'recent_signals': [h.direction for h in recent],
}
def _add_position_with_price_check(
self, symbol: str, tf: TimeFrame, price: float,
stop_loss: float, take_profit: float, reasoning: str,
entry_levels: List[Dict] = None
) -> Optional[Dict]:
"""带价格检查的加仓 - 支持信号中的 entry_levels
Args:
entry_levels: LLM信号中的多级进场价位列表
[{'price': 90000, 'ratio': 0.4, 'level': 0}, ...]
"""
account = self.accounts[symbol][tf]
pos = account.position
if not pos or pos.side == 'FLAT':
return None
# 检查是否已达最大层级
current_level = pos.pyramid_level
if current_level >= len(PYRAMID_LEVELS):
return None
# ========== 优化：使用信号中的 entry_levels ==========
if entry_levels and len(entry_levels) > current_level:
# 获取当前应该的加仓价位
target_entry = entry_levels[current_level]
target_price = target_entry.get('price', 0)
if target_price > 0:
# 检查当前价格是否到达目标加仓价位
price_tolerance = 0.002 # 0.2% 容差
if pos.side == 'LONG':
# 做多：当前价格需要 ≤ 目标价格 (价格下跌才加仓)
if price > target_price * (1 + price_tolerance):
return {
'skipped': True,
'reason': f'未触及加仓价位 L{current_level+1}',
'current_price': price,
'target_price': target_price,
'next_level': current_level + 1,
'entry_levels': entry_levels,
}
else: # SHORT
# 做空：当前价格需要 ≥ 目标价格 (价格上涨才加仓)
if price < target_price * (1 - price_tolerance):
return {
'skipped': True,
'reason': f'未触及加仓价位 L{current_level+1}',
'current_price': price,
'target_price': target_price,
'next_level': current_level + 1,
'entry_levels': entry_levels,
}
# 价格到达目标价位，执行加仓
logger.info(
f"[{symbol}] 触及加仓价位 L{current_level+1}: "
f"目标=${target_price:.2f}, 当前=${price:.2f}"
)
return self._add_position(symbol, tf, price, stop_loss, take_profit, reasoning)
# ========== 回退：使用均价改善检查 ==========
avg_price = pos.entry_price
improvement_required = PYRAMID_PRICE_IMPROVEMENT
if pos.side == 'LONG':
# 做多：加仓价格需要比均价低
price_improvement = (avg_price - price) / avg_price
if price_improvement < improvement_required:
return {
'skipped': True,
'reason': f'加仓价格未改善: 需低于均价{improvement_required*100:.1f}%',
'avg_price': avg_price,
'current_price': price,
'improvement_pct': price_improvement * 100,
'required_improvement_pct': improvement_required * 100,
}
else: # SHORT
# 做空：加仓价格需要比均价高
price_improvement = (price - avg_price) / avg_price
if price_improvement < improvement_required:
return {
'skipped': True,
'reason': f'加仓价格未改善: 需高于均价{improvement_required*100:.1f}%',
'avg_price': avg_price,
'current_price': price,
'improvement_pct': price_improvement * 100,
'required_improvement_pct': improvement_required * 100,
}
# 价格检查通过，执行加仓
return self._add_position(symbol, tf, price, stop_loss, take_profit, reasoning)
def _adjust_stop_loss_take_profit(
self, direction: str, entry_price: float,
signal_sl: float, signal_tp: float,
atr: float, config: Dict
) -> tuple:
"""调整止损止盈（基于ATR验证合理性）"""
if atr <= 0:
return signal_sl, signal_tp
# 计算最小止损距离 (1.5 ATR)
min_sl_distance = atr * 1.5
# 计算当前止损距离
if direction == 'LONG':
current_sl_distance = entry_price - signal_sl
# 如果止损太近，调整
if current_sl_distance < min_sl_distance:
adjusted_sl = entry_price - min_sl_distance
logger.info(f"止损调整: ${signal_sl:.2f} -> ${adjusted_sl:.2f} (基于ATR)")
signal_sl = adjusted_sl
else: # SHORT
current_sl_distance = signal_sl - entry_price
if current_sl_distance < min_sl_distance:
adjusted_sl = entry_price + min_sl_distance
logger.info(f"止损调整: ${signal_sl:.2f} -> ${adjusted_sl:.2f} (基于ATR)")
signal_sl = adjusted_sl
return signal_sl, signal_tp
def _extract_timeframe_signal(
self, signal: Dict[str, Any], signal_keys: List[str]
) -> Optional[Dict[str, Any]]: