mike
e23f8d38c0
方法 2 (百分比波动率) + 方法 3 (关键价位) 混合算法 feat(intraday-regime-detector): 新 skill - 日内市场状态判别 来源: DeepSeek chat share 26iikphv8h94feze9q 核心: R² + ADF + 历史波动率, 识别趋势市 / 震荡市 / 混乱 推荐: 趋势跟踪 / 网格交易 / 布林带回归 / NO_TRADE
292 lines
10 KiB
Python
292 lines
10 KiB
Python
"""
|
||
exit_levels.py - 港美股日内做T 出场点位计算 (混合公式)
|
||
|
||
设计: 方法 2 (百分比波动率) + 方法 3 (关键价位) 混合
|
||
- 关键位: day_high/low / prev_high/low / VWAP (从 indicators.py)
|
||
- 波动率: ATR (从 indicators.py)
|
||
- R:R 强制下限 1.5, 不达标信号否决
|
||
|
||
⚠️ 这是港美股做T专用, 币圈用 crypto-t-monitor 的 ATR 公式 (独立)
|
||
|
||
用法:
|
||
from exit_levels import calc_exit_levels
|
||
|
||
result = calc_exit_levels(
|
||
entry=100.0,
|
||
atr=5.0,
|
||
current_price=100.0,
|
||
day_high=103.0,
|
||
day_low=97.0,
|
||
prev_high=106.0,
|
||
prev_low=94.0,
|
||
vwap=101.0,
|
||
side='long',
|
||
min_rr=1.5,
|
||
)
|
||
if result is None:
|
||
print("信号否决: R:R 不达标")
|
||
else:
|
||
sl, tp1, tp2 = result
|
||
print(f"SL={sl} TP1={tp1} TP2={tp2}")
|
||
"""
|
||
|
||
from dataclasses import dataclass
|
||
from typing import Optional, Literal
|
||
|
||
|
||
@dataclass
|
||
class ExitLevels:
|
||
"""出场点位结果"""
|
||
sl: float # 止损价位
|
||
tp1: float # 第一止盈 (半平)
|
||
tp2: float # 第二止盈 (全平)
|
||
entry: float # 入场价 (回填, 方便调用方记录)
|
||
side: str # 'long' / 'short'
|
||
risk: float # 风险 (entry - SL)
|
||
reward: float # 奖励 (TP1 - entry)
|
||
rr_ratio: float # R:R (reward/risk)
|
||
sl_method: str # 'vol_pct' | 'key_level' | 'hybrid'
|
||
tp_method: str # 'vol_pct' | 'key_level' | 'hybrid'
|
||
note: str = "" # 备注 (VWAP 锁定等)
|
||
|
||
|
||
def calc_exit_levels(
|
||
entry: float,
|
||
atr: float,
|
||
current_price: float,
|
||
day_high: Optional[float] = None,
|
||
day_low: Optional[float] = None,
|
||
prev_high: Optional[float] = None,
|
||
prev_low: Optional[float] = None,
|
||
vwap: Optional[float] = None,
|
||
side: Literal['long', 'short'] = 'long',
|
||
min_rr: float = 1.5,
|
||
# 方法 2 权重 (百分比波动率)
|
||
vol_sl_multi: float = 1.0, # SL 距离 = vol × sl_multi
|
||
vol_tp1_multi: float = 2.0, # TP1 距离 = vol × tp1_multi (默认 2.0 倍 SL → R:R 2:1)
|
||
vol_tp2_multi: float = 3.0, # TP2 距离 = vol × tp2_multi
|
||
# 方法 3 权重 (关键位) - 离关键位的 buffer
|
||
key_level_buffer_pct: float = 0.001, # 0.1% 缓冲 (避免瞬时触发)
|
||
) -> Optional[ExitLevels]:
|
||
"""
|
||
计算 SL/TP1/TP2 (混合公式: 波动率 + 关键位)
|
||
|
||
Args:
|
||
entry: 入场价 (假设已知, 或用 adjust_to_ask1 拿到的价)
|
||
atr: 当前 K 线 ATR (14 周期, 从 indicators.atr())
|
||
current_price: 当前实时价 (用于 vol_pct 计算)
|
||
day_high/low: 今日最高/最低 (从 longbridge quote)
|
||
prev_high/low: 昨日最高/最低 (从 longbridge K 线)
|
||
vwap: 成交量加权平均价 (从 indicators.vwap())
|
||
side: 'long' (做多) / 'short' (做空)
|
||
min_rr: 最小 R:R (默认 1.5)
|
||
vol_sl_multi / tp1_multi / tp2_multi: ATR 倍数
|
||
key_level_buffer_pct: 关键位 buffer (避免价格精确等于关键位)
|
||
|
||
Returns:
|
||
ExitLevels 或 None (R:R 不达标)
|
||
|
||
Examples:
|
||
>>> calc_exit_levels(entry=100, atr=5, current_price=100,
|
||
... day_high=115, day_low=92, prev_high=118, prev_low=90,
|
||
... vwap=105, side='long')
|
||
ExitLevels(sl=91.2, tp1=114.2, tp2=120.0, ...)
|
||
"""
|
||
if entry <= 0 or atr <= 0 or current_price <= 0:
|
||
raise ValueError("entry/atr/current_price must be > 0")
|
||
|
||
# === 方法 2: 百分比波动率 (基于 ATR) ===
|
||
vol_pct = atr / current_price
|
||
vol_sl_dist = vol_pct * vol_sl_multi
|
||
vol_tp1_dist = vol_pct * vol_tp1_multi
|
||
vol_tp2_dist = vol_pct * vol_tp2_multi
|
||
|
||
if side == 'long':
|
||
sl_vol = entry * (1 - vol_sl_dist)
|
||
tp1_vol = entry * (1 + vol_tp1_dist)
|
||
tp2_vol = entry * (1 + vol_tp2_dist)
|
||
else:
|
||
sl_vol = entry * (1 + vol_sl_dist)
|
||
tp1_vol = entry * (1 - vol_tp1_dist)
|
||
tp2_vol = entry * (1 - vol_tp2_dist)
|
||
|
||
# === 方法 3: 关键价位 ===
|
||
# 多仓: SL 取 entry **下方**的支撑; TP1 取 entry **上方**的阻力
|
||
# 空仓: SL 取 entry **上方**的阻力; TP1 取 entry **下方**的支撑
|
||
sl_key = None
|
||
tp1_key = None
|
||
note = ""
|
||
|
||
if side == 'long':
|
||
# SL 关键位 (entry 下方)
|
||
sl_candidates = []
|
||
if day_low is not None and day_low < entry:
|
||
sl_candidates.append(day_low * (1 - key_level_buffer_pct))
|
||
if prev_low is not None and prev_low < entry:
|
||
sl_candidates.append(prev_low * (1 - key_level_buffer_pct))
|
||
if vwap is not None and vwap < entry:
|
||
sl_candidates.append(vwap * (1 - key_level_buffer_pct))
|
||
sl_key = max(sl_candidates) if sl_candidates else None
|
||
|
||
# TP1 关键位 (entry 上方)
|
||
tp1_candidates = []
|
||
if day_high is not None and day_high > entry:
|
||
tp1_candidates.append(day_high * (1 - key_level_buffer_pct))
|
||
if prev_high is not None and prev_high > entry:
|
||
tp1_candidates.append(prev_high * (1 - key_level_buffer_pct))
|
||
if vwap is not None and vwap > entry:
|
||
tp1_candidates.append(vwap * (1 - key_level_buffer_pct))
|
||
tp1_key = min(tp1_candidates) if tp1_candidates else None
|
||
else: # short
|
||
# SL 关键位 (entry 上方, 空仓止损 = 价格涨到这里平)
|
||
sl_candidates = []
|
||
if day_high is not None and day_high > entry:
|
||
sl_candidates.append(day_high * (1 + key_level_buffer_pct))
|
||
if prev_high is not None and prev_high > entry:
|
||
sl_candidates.append(prev_high * (1 + key_level_buffer_pct))
|
||
if vwap is not None and vwap > entry:
|
||
sl_candidates.append(vwap * (1 + key_level_buffer_pct))
|
||
sl_key = min(sl_candidates) if sl_candidates else None
|
||
|
||
# TP1 关键位 (entry 下方)
|
||
tp1_candidates = []
|
||
if day_low is not None and day_low < entry:
|
||
tp1_candidates.append(day_low * (1 + key_level_buffer_pct))
|
||
if prev_low is not None and prev_low < entry:
|
||
tp1_candidates.append(prev_low * (1 + key_level_buffer_pct))
|
||
if vwap is not None and vwap < entry:
|
||
tp1_candidates.append(vwap * (1 + key_level_buffer_pct))
|
||
tp1_key = max(tp1_candidates) if tp1_candidates else None
|
||
|
||
# === 混合: vol_pct 主导 (70%), 关键位微调 (30%) ===
|
||
# 关键位 30% 权重, 防止 VWAP 等动态位锁死
|
||
# vol_pct 至少占 70% (最终值不会偏离 vol_pct 太远)
|
||
if side == 'long':
|
||
if sl_key is not None:
|
||
SL = sl_vol * 0.7 + sl_key * 0.3
|
||
sl_method = 'hybrid_blend'
|
||
else:
|
||
SL = sl_vol
|
||
sl_method = 'vol_pct'
|
||
|
||
if tp1_key is not None:
|
||
TP1 = tp1_vol * 0.7 + tp1_key * 0.3
|
||
tp_method = 'hybrid_blend'
|
||
else:
|
||
TP1 = tp1_vol
|
||
tp_method = 'vol_pct'
|
||
|
||
TP2 = tp2_vol
|
||
else: # short
|
||
if sl_key is not None:
|
||
SL = sl_vol * 0.7 + sl_key * 0.3
|
||
sl_method = 'hybrid_blend'
|
||
else:
|
||
SL = sl_vol
|
||
sl_method = 'vol_pct'
|
||
|
||
if tp1_key is not None:
|
||
TP1 = tp1_vol * 0.7 + tp1_key * 0.3
|
||
tp_method = 'hybrid_blend'
|
||
else:
|
||
TP1 = tp1_vol
|
||
tp_method = 'vol_pct'
|
||
|
||
TP2 = tp2_vol
|
||
|
||
# === R:R 检查 ===
|
||
if side == 'long':
|
||
risk = entry - SL
|
||
reward = TP1 - entry
|
||
else:
|
||
risk = SL - entry
|
||
reward = entry - TP1
|
||
|
||
if risk <= 0:
|
||
return None # 止损 >= 入场 (逻辑错误)
|
||
|
||
rr_ratio = reward / risk if risk > 0 else 0
|
||
|
||
# 风险 vs reward
|
||
if abs(reward) < min_rr * abs(risk):
|
||
# R:R 不达标
|
||
if sl_key == tp1_key and sl_key is not None:
|
||
note = f"VWAP 既作支撑又作阻力, 价格窄幅震荡 (SL=TP1={sl_key:.2f})"
|
||
else:
|
||
note = f"R:R {rr_ratio:.2f} < {min_rr}, 信号否决"
|
||
return None
|
||
|
||
return ExitLevels(
|
||
sl=SL,
|
||
tp1=TP1,
|
||
tp2=TP2,
|
||
entry=entry,
|
||
side=side,
|
||
risk=abs(risk),
|
||
reward=abs(reward),
|
||
rr_ratio=rr_ratio,
|
||
sl_method=sl_method,
|
||
tp_method=tp_method,
|
||
note=note,
|
||
)
|
||
|
||
|
||
def format_levels(levels: ExitLevels) -> str:
|
||
"""格式化输出 (QQ 推送用)"""
|
||
side_emoji = '🟢' if levels.side == 'long' else '🔴'
|
||
return (
|
||
f"{side_emoji} {levels.side.upper()} @ ${levels.entry:.2f}\n"
|
||
f" SL: ${levels.sl:.2f} ({levels.sl_method})\n"
|
||
f" TP1: ${levels.tp1:.2f} ({levels.tp_method})\n"
|
||
f" TP2: ${levels.tp2:.2f}\n"
|
||
f" Risk/Reward: 1:{levels.rr_ratio:.2f}"
|
||
)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# 自检: 用 indicators.py 的真实数据测试
|
||
print("=" * 60)
|
||
print("exit_levels.py - 自检")
|
||
print("=" * 60)
|
||
|
||
# 案例 1: NVDA 高波动 (有 R:R)
|
||
print("\n[案例 1] NVDA $100, ATR=$8, day H/L=$115/$92, prev H/L=$118/$90")
|
||
result = calc_exit_levels(
|
||
entry=100.0, atr=8.0, current_price=100.0,
|
||
day_high=115.0, day_low=92.0,
|
||
prev_high=118.0, prev_low=90.0,
|
||
vwap=105.0,
|
||
side='long',
|
||
)
|
||
if result:
|
||
print(format_levels(result))
|
||
else:
|
||
print("❌ 信号否决")
|
||
|
||
# 案例 2: 价在 VWAP 上下窄幅震荡
|
||
print("\n[案例 2] NVDA $100, ATR=$3, VWAP=$101 (紧贴)")
|
||
result = calc_exit_levels(
|
||
entry=100.0, atr=3.0, current_price=100.0,
|
||
day_high=103.0, day_low=98.0,
|
||
prev_high=105.0, prev_low=95.0,
|
||
vwap=101.0,
|
||
side='long',
|
||
)
|
||
if result:
|
||
print(format_levels(result))
|
||
else:
|
||
print("❌ 信号否决 (R:R 不达标 / VWAP 锁定)")
|
||
|
||
# 案例 3: 空仓 + 大波动
|
||
print("\n[案例 3] TSDA short $200, ATR=$12")
|
||
result = calc_exit_levels(
|
||
entry=200.0, atr=12.0, current_price=200.0,
|
||
day_high=212.0, day_low=188.0,
|
||
prev_high=215.0, prev_low=185.0,
|
||
vwap=205.0,
|
||
side='short',
|
||
)
|
||
if result:
|
||
print(format_levels(result))
|
||
else:
|
||
print("❌ 信号否决") |