316 lines
12 KiB
Python
316 lines
12 KiB
Python
"""
|
||
ATR策略模块 - 基于平均真实波幅(ATR)的动态止损止盈策略
|
||
"""
|
||
import logging
|
||
from typing import Optional, Dict, Tuple
|
||
try:
|
||
from . import config
|
||
from .indicators import TechnicalIndicators
|
||
except ImportError:
|
||
import config
|
||
from indicators import TechnicalIndicators
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class ATRStrategy:
|
||
"""ATR策略类 - 基于ATR的动态止损止盈计算"""
|
||
|
||
def __init__(self):
|
||
"""初始化ATR策略"""
|
||
self.use_atr = config.TRADING_CONFIG.get('USE_ATR_STOP_LOSS', True)
|
||
self.atr_sl_multiplier = config.TRADING_CONFIG.get('ATR_STOP_LOSS_MULTIPLIER', 1.8)
|
||
self.atr_tp_multiplier = config.TRADING_CONFIG.get('ATR_TAKE_PROFIT_MULTIPLIER', 3.0)
|
||
self.risk_reward_ratio = config.TRADING_CONFIG.get('RISK_REWARD_RATIO', 3.0)
|
||
self.atr_period = config.TRADING_CONFIG.get('ATR_PERIOD', 14)
|
||
|
||
# 动态调整ATR倍数的参数
|
||
self.use_dynamic_atr_multiplier = config.TRADING_CONFIG.get('USE_DYNAMIC_ATR_MULTIPLIER', False)
|
||
self.atr_multiplier_min = config.TRADING_CONFIG.get('ATR_MULTIPLIER_MIN', 1.5)
|
||
self.atr_multiplier_max = config.TRADING_CONFIG.get('ATR_MULTIPLIER_MAX', 2.5)
|
||
|
||
def calculate_stop_loss(
|
||
self,
|
||
entry_price: float,
|
||
side: str,
|
||
atr: Optional[float] = None,
|
||
atr_percent: Optional[float] = None,
|
||
volatility: Optional[float] = None
|
||
) -> Tuple[Optional[float], Optional[float], Dict]:
|
||
"""
|
||
计算基于ATR的止损价格
|
||
|
||
Args:
|
||
entry_price: 入场价格
|
||
side: 方向 'BUY' 或 'SELL'
|
||
atr: ATR绝对值（可选，如果提供则直接使用）
|
||
atr_percent: ATR百分比（可选，如果提供则直接使用，否则从atr计算）
|
||
volatility: 市场波动率（可选，用于动态调整ATR倍数）
|
||
|
||
Returns:
|
||
(止损价格, 止损距离, 详细信息字典)
|
||
"""
|
||
if not self.use_atr:
|
||
return None, None, {}
|
||
|
||
# 如果没有提供ATR，无法计算
|
||
if atr is None and atr_percent is None:
|
||
logger.debug("ATR不可用，无法计算ATR止损")
|
||
return None, None, {}
|
||
|
||
# 计算ATR百分比（如果未提供）
|
||
if atr_percent is None and atr is not None and entry_price > 0:
|
||
atr_percent = atr / entry_price
|
||
|
||
if atr_percent is None or atr_percent <= 0:
|
||
logger.debug("ATR百分比无效，无法计算ATR止损")
|
||
return None, None, {}
|
||
|
||
# 动态调整ATR倍数（根据市场波动率）
|
||
if self.use_dynamic_atr_multiplier and volatility is not None:
|
||
# 波动率越高，ATR倍数越大（更宽松的止损）
|
||
# 波动率范围假设：0.01-0.10 (1%-10%)
|
||
volatility_factor = min(max((volatility - 0.01) / 0.09, 0), 1) # 归一化到0-1
|
||
dynamic_multiplier = (
|
||
self.atr_multiplier_min +
|
||
(self.atr_multiplier_max - self.atr_multiplier_min) * volatility_factor
|
||
)
|
||
atr_multiplier = dynamic_multiplier
|
||
logger.debug(f"动态ATR倍数: 波动率={volatility:.4f}, 倍数={atr_multiplier:.2f}")
|
||
else:
|
||
atr_multiplier = self.atr_sl_multiplier
|
||
|
||
# 计算止损距离（基于ATR百分比）
|
||
stop_distance_percent = atr_percent * atr_multiplier
|
||
|
||
# 计算止损价格
|
||
if side == 'BUY': # 做多，止损价低于入场价
|
||
stop_loss_price = entry_price * (1 - stop_distance_percent)
|
||
stop_distance = entry_price - stop_loss_price
|
||
else: # 做空，止损价高于入场价
|
||
stop_loss_price = entry_price * (1 + stop_distance_percent)
|
||
stop_distance = stop_loss_price - entry_price
|
||
|
||
details = {
|
||
'atr': atr,
|
||
'atr_percent': atr_percent,
|
||
'atr_multiplier': atr_multiplier,
|
||
'stop_distance_percent': stop_distance_percent,
|
||
'stop_distance': stop_distance,
|
||
'method': 'ATR动态止损'
|
||
}
|
||
|
||
# 格式化ATR显示
|
||
atr_display = f"{atr:.4f}" if atr is not None and atr > 0 else "N/A"
|
||
|
||
logger.info(
|
||
f"ATR止损计算 ({side}): "
|
||
f"ATR={atr_display}, "
|
||
f"ATR%={atr_percent*100:.2f}%, "
|
||
f"倍数={atr_multiplier:.2f}, "
|
||
f"止损距离={stop_distance_percent*100:.2f}%, "
|
||
f"止损价={stop_loss_price:.4f}"
|
||
)
|
||
|
||
return stop_loss_price, stop_distance, details
|
||
|
||
def calculate_take_profit(
|
||
self,
|
||
entry_price: float,
|
||
side: str,
|
||
stop_distance: Optional[float] = None,
|
||
atr: Optional[float] = None,
|
||
atr_percent: Optional[float] = None,
|
||
use_risk_reward_ratio: bool = True
|
||
) -> Tuple[Optional[float], Optional[float], Dict]:
|
||
"""
|
||
计算基于ATR的止盈价格
|
||
|
||
Args:
|
||
entry_price: 入场价格
|
||
side: 方向 'BUY' 或 'SELL'
|
||
stop_distance: 止损距离（价格差，可选，如果提供则基于盈亏比计算）
|
||
atr: ATR绝对值（可选）
|
||
atr_percent: ATR百分比（可选）
|
||
use_risk_reward_ratio: 是否使用盈亏比（True：基于止损距离×盈亏比，False：基于ATR倍数）
|
||
|
||
Returns:
|
||
(止盈价格, 止盈距离, 详细信息字典)
|
||
"""
|
||
if not self.use_atr:
|
||
return None, None, {}
|
||
|
||
# 方法1：基于止损距离和盈亏比计算（优先）
|
||
if use_risk_reward_ratio and stop_distance is not None and stop_distance > 0:
|
||
take_profit_distance = stop_distance * self.risk_reward_ratio
|
||
|
||
if side == 'BUY': # 做多，止盈价高于入场价
|
||
take_profit_price = entry_price + take_profit_distance
|
||
else: # 做空，止盈价低于入场价
|
||
take_profit_price = entry_price - take_profit_distance
|
||
|
||
details = {
|
||
'stop_distance': stop_distance,
|
||
'risk_reward_ratio': self.risk_reward_ratio,
|
||
'take_profit_distance': take_profit_distance,
|
||
'method': 'ATR止损距离×盈亏比'
|
||
}
|
||
|
||
logger.info(
|
||
f"ATR止盈计算 ({side}, 基于盈亏比): "
|
||
f"止损距离={stop_distance:.4f}, "
|
||
f"盈亏比={self.risk_reward_ratio:.1f}, "
|
||
f"止盈距离={take_profit_distance:.4f}, "
|
||
f"止盈价={take_profit_price:.4f}"
|
||
)
|
||
|
||
return take_profit_price, take_profit_distance, details
|
||
|
||
# 方法2：基于ATR倍数计算（备选）
|
||
if atr is None and atr_percent is None:
|
||
logger.debug("ATR不可用，无法计算ATR止盈")
|
||
return None, None, {}
|
||
|
||
# 计算ATR百分比（如果未提供）
|
||
if atr_percent is None and atr is not None and entry_price > 0:
|
||
atr_percent = atr / entry_price
|
||
|
||
if atr_percent is None or atr_percent <= 0:
|
||
logger.debug("ATR百分比无效，无法计算ATR止盈")
|
||
return None, None, {}
|
||
|
||
# 计算止盈距离（基于ATR百分比）
|
||
take_profit_distance_percent = atr_percent * self.atr_tp_multiplier
|
||
|
||
# 计算止盈价格
|
||
if side == 'BUY': # 做多，止盈价高于入场价
|
||
take_profit_price = entry_price * (1 + take_profit_distance_percent)
|
||
take_profit_distance = take_profit_price - entry_price
|
||
else: # 做空，止盈价低于入场价
|
||
take_profit_price = entry_price * (1 - take_profit_distance_percent)
|
||
take_profit_distance = entry_price - take_profit_price
|
||
|
||
details = {
|
||
'atr': atr,
|
||
'atr_percent': atr_percent,
|
||
'atr_multiplier': self.atr_tp_multiplier,
|
||
'take_profit_distance_percent': take_profit_distance_percent,
|
||
'take_profit_distance': take_profit_distance,
|
||
'method': 'ATR倍数止盈'
|
||
}
|
||
|
||
# 格式化ATR显示
|
||
atr_display = f"{atr:.4f}" if atr is not None and atr > 0 else "N/A"
|
||
|
||
logger.info(
|
||
f"ATR止盈计算 ({side}, 基于ATR倍数): "
|
||
f"ATR={atr_display}, "
|
||
f"ATR%={atr_percent*100:.2f}%, "
|
||
f"倍数={self.atr_tp_multiplier:.2f}, "
|
||
f"止盈距离={take_profit_distance_percent*100:.2f}%, "
|
||
f"止盈价={take_profit_price:.4f}"
|
||
)
|
||
|
||
return take_profit_price, take_profit_distance, details
|
||
|
||
def calculate_atr_levels(
|
||
self,
|
||
entry_price: float,
|
||
side: str,
|
||
atr: Optional[float] = None,
|
||
atr_percent: Optional[float] = None,
|
||
volatility: Optional[float] = None
|
||
) -> Dict:
|
||
"""
|
||
计算完整的ATR止损止盈方案
|
||
|
||
Args:
|
||
entry_price: 入场价格
|
||
side: 方向 'BUY' 或 'SELL'
|
||
atr: ATR绝对值（可选）
|
||
atr_percent: ATR百分比（可选）
|
||
volatility: 市场波动率（可选）
|
||
|
||
Returns:
|
||
包含止损、止盈等信息的字典
|
||
"""
|
||
result = {
|
||
'use_atr': self.use_atr,
|
||
'entry_price': entry_price,
|
||
'side': side,
|
||
'stop_loss_price': None,
|
||
'stop_loss_distance': None,
|
||
'take_profit_price': None,
|
||
'take_profit_distance': None,
|
||
'risk_reward_ratio': None,
|
||
'details': {}
|
||
}
|
||
|
||
if not self.use_atr:
|
||
return result
|
||
|
||
# 计算止损
|
||
stop_loss_price, stop_distance, stop_details = self.calculate_stop_loss(
|
||
entry_price, side, atr, atr_percent, volatility
|
||
)
|
||
|
||
if stop_loss_price is None:
|
||
return result
|
||
|
||
result['stop_loss_price'] = stop_loss_price
|
||
result['stop_loss_distance'] = stop_distance
|
||
result['details']['stop_loss'] = stop_details
|
||
|
||
# 计算止盈（基于止损距离和盈亏比）
|
||
take_profit_price, take_profit_distance, tp_details = self.calculate_take_profit(
|
||
entry_price, side, stop_distance, atr, atr_percent, use_risk_reward_ratio=True
|
||
)
|
||
|
||
if take_profit_price is not None:
|
||
result['take_profit_price'] = take_profit_price
|
||
result['take_profit_distance'] = take_profit_distance
|
||
result['risk_reward_ratio'] = (
|
||
take_profit_distance / stop_distance if stop_distance > 0 else None
|
||
)
|
||
result['details']['take_profit'] = tp_details
|
||
|
||
return result
|
||
|
||
@staticmethod
|
||
def calculate_atr_from_klines(
|
||
klines: list,
|
||
period: int = 14
|
||
) -> Tuple[Optional[float], Optional[float], Optional[float]]:
|
||
"""
|
||
从K线数据计算ATR和ATR百分比
|
||
|
||
Args:
|
||
klines: K线数据列表，格式为 [open, high, low, close, ...]
|
||
period: ATR计算周期，默认14
|
||
|
||
Returns:
|
||
(ATR绝对值, ATR百分比, 当前价格)
|
||
"""
|
||
if not klines or len(klines) < period + 1:
|
||
return None, None, None
|
||
|
||
try:
|
||
high_prices = [float(k[2]) for k in klines] # high
|
||
low_prices = [float(k[3]) for k in klines] # low
|
||
close_prices = [float(k[4]) for k in klines] # close
|
||
|
||
current_price = close_prices[-1] if close_prices else None
|
||
|
||
atr = TechnicalIndicators.calculate_atr(high_prices, low_prices, close_prices, period)
|
||
|
||
if atr is None or current_price is None or current_price <= 0:
|
||
return None, None, None
|
||
|
||
atr_percent = atr / current_price
|
||
|
||
return atr, atr_percent, current_price
|
||
|
||
except (IndexError, ValueError, TypeError) as e:
|
||
logger.debug(f"从K线计算ATR失败: {e}")
|
||
return None, None, None
|