396 lines
14 KiB
Python
396 lines
14 KiB
Python
"""
|
||
PriceActionStrategy - 纯价格行为策略
|
||
|
||
核心原则：
|
||
零指标。不用 EMA、RSI、MACD、ATR 或任何计算指标。
|
||
只看 K线本身（Open/High/Low/Close/Volume）。
|
||
|
||
价格行为判断方法：
|
||
1. 市场结构（趋势）：用 Swing High / Swing Low 判断
|
||
- 上升趋势 = Higher High + Higher Low
|
||
- 下降趋势 = Lower High + Lower Low
|
||
2. 入场信号：纯K线形态
|
||
- Pin Bar（锤子线/射击之星）
|
||
- 吞没形态（Engulfing）
|
||
- Inside Bar 突破
|
||
3. 出场：用前一个 Swing High/Low 作为止盈目标
|
||
4. 止损：放在信号K线的另一端
|
||
|
||
使用时间框架：
|
||
- 5m（主时间框架）：入场/出场 + 结构判断
|
||
"""
|
||
|
||
import numpy as np
|
||
from pandas import DataFrame
|
||
|
||
from freqtrade.strategy import IStrategy
|
||
|
||
|
||
class PriceActionStrategy(IStrategy):
|
||
"""
|
||
纯价格行为策略 - 零指标
|
||
"""
|
||
|
||
INTERFACE_VERSION = 3
|
||
|
||
# === 基础配置 ===
|
||
timeframe = "5m"
|
||
can_short = True
|
||
stoploss = -0.03 # 3% 硬止损安全网
|
||
trailing_stop = False
|
||
use_custom_stoploss = False
|
||
|
||
startup_candle_count: int = 100
|
||
|
||
# 不用 ROI 自动止盈，让价格行为决定出场
|
||
minimal_roi = {}
|
||
|
||
order_types = {
|
||
"entry": "market",
|
||
"exit": "market",
|
||
"stoploss": "market",
|
||
"stoploss_on_exchange": False,
|
||
}
|
||
|
||
use_exit_signal = True
|
||
exit_profit_only = False
|
||
ignore_roi_if_entry_signal = False
|
||
|
||
# === 参数 ===
|
||
swing_lookback = 10 # Swing High/Low 回看K线数
|
||
min_body_ratio = 0.55 # 最小实体占比（实体/全幅）
|
||
pin_shadow_ratio = 2.5 # Pin Bar 影线至少是实体的 N 倍
|
||
engulf_body_ratio = 1.2 # 吞没K线实体至少是前一根的 N 倍
|
||
|
||
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
"""
|
||
纯价格行为 — 只从 OHLCV 提取结构信息，不计算任何技术指标。
|
||
"""
|
||
df = dataframe
|
||
|
||
# ========== K线基础属性 ==========
|
||
df["body"] = abs(df["close"] - df["open"])
|
||
df["candle_range"] = df["high"] - df["low"]
|
||
df["body_ratio"] = df["body"] / (df["candle_range"] + 1e-10)
|
||
df["upper_shadow"] = df["high"] - df[["close", "open"]].max(axis=1)
|
||
df["lower_shadow"] = df[["close", "open"]].min(axis=1) - df["low"]
|
||
df["is_bull"] = (df["close"] > df["open"]).astype(int)
|
||
df["is_bear"] = (df["close"] < df["open"]).astype(int)
|
||
|
||
# ========== Swing High / Swing Low ==========
|
||
# Swing High: 当前 high 是前后 N 根K线中最高的
|
||
# Swing Low: 当前 low 是前后 N 根K线中最低的
|
||
n = self.swing_lookback
|
||
df["swing_high"] = df["high"].rolling(window=2 * n + 1, center=True).apply(
|
||
lambda x: 1 if x.iloc[n] == x.max() else 0, raw=False
|
||
)
|
||
df["swing_low"] = df["low"].rolling(window=2 * n + 1, center=True).apply(
|
||
lambda x: 1 if x.iloc[n] == x.min() else 0, raw=False
|
||
)
|
||
|
||
# 记录最近的 Swing High/Low 价格
|
||
df["last_swing_high"] = np.nan
|
||
df["last_swing_low"] = np.nan
|
||
df["prev_swing_high"] = np.nan
|
||
df["prev_swing_low"] = np.nan
|
||
|
||
df.loc[df["swing_high"] == 1, "last_swing_high"] = df["high"]
|
||
df["last_swing_high"] = df["last_swing_high"].ffill()
|
||
|
||
df.loc[df["swing_low"] == 1, "last_swing_low"] = df["low"]
|
||
df["last_swing_low"] = df["last_swing_low"].ffill()
|
||
|
||
# 前一个 Swing High/Low（用于判断 HH/HL/LH/LL）
|
||
swing_high_prices = df.loc[df["swing_high"] == 1, "high"]
|
||
swing_low_prices = df.loc[df["swing_low"] == 1, "low"]
|
||
|
||
# 构建 prev_swing_high: 每个 swing high 点对应的上一个 swing high
|
||
sh_idx = swing_high_prices.index.tolist()
|
||
for i in range(1, len(sh_idx)):
|
||
df.loc[sh_idx[i], "prev_swing_high"] = swing_high_prices.loc[sh_idx[i - 1]]
|
||
df["prev_swing_high"] = df["prev_swing_high"].ffill()
|
||
|
||
sl_idx = swing_low_prices.index.tolist()
|
||
for i in range(1, len(sl_idx)):
|
||
df.loc[sl_idx[i], "prev_swing_low"] = swing_low_prices.loc[sl_idx[i - 1]]
|
||
df["prev_swing_low"] = df["prev_swing_low"].ffill()
|
||
|
||
# ========== 市场结构（趋势）==========
|
||
# Higher High + Higher Low = 上升趋势
|
||
# Lower High + Lower Low = 下降趋势
|
||
df["higher_high"] = (df["last_swing_high"] > df["prev_swing_high"]).astype(int)
|
||
df["higher_low"] = (df["last_swing_low"] > df["prev_swing_low"]).astype(int)
|
||
df["lower_high"] = (df["last_swing_high"] < df["prev_swing_high"]).astype(int)
|
||
df["lower_low"] = (df["last_swing_low"] < df["prev_swing_low"]).astype(int)
|
||
|
||
df["uptrend"] = ((df["higher_high"] == 1) & (df["higher_low"] == 1)).astype(int)
|
||
df["downtrend"] = ((df["lower_high"] == 1) & (df["lower_low"] == 1)).astype(int)
|
||
|
||
# ========== 价格行为形态 ==========
|
||
|
||
# --- Pin Bar（锤子线 / 射击之星）---
|
||
# 看涨 Pin Bar: 长下影线，短上影线，实体在上半部分
|
||
df["bullish_pin"] = (
|
||
(df["lower_shadow"] > df["body"] * self.pin_shadow_ratio)
|
||
& (df["lower_shadow"] > df["upper_shadow"] * 2)
|
||
& (df["body_ratio"] > 0.15) # 不是十字星
|
||
& (df["is_bull"] == 1)
|
||
).astype(int)
|
||
|
||
# 看跌 Pin Bar: 长上影线，短下影线，实体在下半部分
|
||
df["bearish_pin"] = (
|
||
(df["upper_shadow"] > df["body"] * self.pin_shadow_ratio)
|
||
& (df["upper_shadow"] > df["lower_shadow"] * 2)
|
||
& (df["body_ratio"] > 0.15)
|
||
& (df["is_bear"] == 1)
|
||
).astype(int)
|
||
|
||
# --- 吞没形态（Engulfing）---
|
||
prev_body = df["body"].shift(1)
|
||
prev_open = df["open"].shift(1)
|
||
prev_close = df["close"].shift(1)
|
||
|
||
# 看涨吞没: 前一根阴线，当前阳线完全包住前一根
|
||
df["bullish_engulf"] = (
|
||
(df["is_bull"] == 1)
|
||
& (prev_close < prev_open) # 前一根是阴线
|
||
& (df["open"] <= prev_close) # 开盘 <= 前收盘（低开或平开）
|
||
& (df["close"] >= prev_open) # 收盘 >= 前开盘（完全吞没）
|
||
& (df["body"] > prev_body * self.engulf_body_ratio) # 实体更大
|
||
).astype(int)
|
||
|
||
# 看跌吞没
|
||
df["bearish_engulf"] = (
|
||
(df["is_bear"] == 1)
|
||
& (prev_close > prev_open) # 前一根是阳线
|
||
& (df["open"] >= prev_close) # 开盘 >= 前收盘
|
||
& (df["close"] <= prev_open) # 收盘 <= 前开盘
|
||
& (df["body"] > prev_body * self.engulf_body_ratio)
|
||
).astype(int)
|
||
|
||
# --- Inside Bar 突破 ---
|
||
# Inside Bar: 当前K线的 high/low 完全在前一根范围内
|
||
prev_high = df["high"].shift(1)
|
||
prev_low = df["low"].shift(1)
|
||
df["inside_bar"] = (
|
||
(df["high"] <= prev_high)
|
||
& (df["low"] >= prev_low)
|
||
).astype(int)
|
||
|
||
# Inside Bar 之后的突破
|
||
# 向上突破: 前一根是 inside bar，当前收盘 > 母线（前两根）的 high
|
||
mother_high = df["high"].shift(2)
|
||
mother_low = df["low"].shift(2)
|
||
df["inside_break_up"] = (
|
||
(df["inside_bar"].shift(1) == 1)
|
||
& (df["close"] > mother_high)
|
||
& (df["is_bull"] == 1)
|
||
).astype(int)
|
||
|
||
df["inside_break_down"] = (
|
||
(df["inside_bar"].shift(1) == 1)
|
||
& (df["close"] < mother_low)
|
||
& (df["is_bear"] == 1)
|
||
).astype(int)
|
||
|
||
# --- 支撑/阻力突破 ---
|
||
# 突破前一个 Swing High（做多）
|
||
df["break_swing_high"] = (
|
||
(df["close"] > df["last_swing_high"].shift(1))
|
||
& (df["close"].shift(1) <= df["last_swing_high"].shift(1))
|
||
& (df["is_bull"] == 1)
|
||
& (df["body_ratio"] > self.min_body_ratio) # 实体饱满（有力度）
|
||
).astype(int)
|
||
|
||
# 跌破前一个 Swing Low（做空）
|
||
df["break_swing_low"] = (
|
||
(df["close"] < df["last_swing_low"].shift(1))
|
||
& (df["close"].shift(1) >= df["last_swing_low"].shift(1))
|
||
& (df["is_bear"] == 1)
|
||
& (df["body_ratio"] > self.min_body_ratio)
|
||
).astype(int)
|
||
|
||
# --- 成交量确认（只用原始 volume 对比，不算均线）---
|
||
# 当前成交量 > 前3根的最大成交量 = 放量
|
||
df["vol_expand"] = (
|
||
df["volume"] > df["volume"].rolling(3).max().shift(1)
|
||
).astype(int)
|
||
|
||
return dataframe
|
||
|
||
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
"""
|
||
入场条件 — 纯价格行为
|
||
|
||
做多条件（满足任一组）：
|
||
A) 上升趋势 + 看涨 Pin Bar（回调到支撑后反弹信号）
|
||
B) 上升趋势 + 看涨吞没（回调后强势反转）
|
||
C) 上升趋势 + Inside Bar 向上突破（蓄力后爆发）
|
||
D) 突破 Swing High + 放量（结构性突破）
|
||
|
||
做空条件（镜像）
|
||
"""
|
||
df = dataframe
|
||
|
||
# ===== 做多 =====
|
||
conditions_long = []
|
||
|
||
# A) 上升趋势 + 看涨 Pin Bar
|
||
conditions_long.append(
|
||
(df["uptrend"] == 1)
|
||
& (df["bullish_pin"] == 1)
|
||
& (df["vol_expand"] == 1)
|
||
)
|
||
|
||
# B) 上升趋势 + 看涨吞没
|
||
conditions_long.append(
|
||
(df["uptrend"] == 1)
|
||
& (df["bullish_engulf"] == 1)
|
||
)
|
||
|
||
# C) 上升趋势 + Inside Bar 向上突破
|
||
conditions_long.append(
|
||
(df["uptrend"] == 1)
|
||
& (df["inside_break_up"] == 1)
|
||
& (df["vol_expand"] == 1)
|
||
)
|
||
|
||
# D) 突破 Swing High + 放量（不需要已确认趋势，突破本身建立趋势）
|
||
conditions_long.append(
|
||
(df["break_swing_high"] == 1)
|
||
& (df["vol_expand"] == 1)
|
||
)
|
||
|
||
if conditions_long:
|
||
import pandas as pd
|
||
combined = pd.concat(conditions_long, axis=1).any(axis=1)
|
||
dataframe.loc[combined, "enter_long"] = 1
|
||
|
||
# ===== 做空 =====
|
||
conditions_short = []
|
||
|
||
# A) 下降趋势 + 看跌 Pin Bar
|
||
conditions_short.append(
|
||
(df["downtrend"] == 1)
|
||
& (df["bearish_pin"] == 1)
|
||
& (df["vol_expand"] == 1)
|
||
)
|
||
|
||
# B) 下降趋势 + 看跌吞没
|
||
conditions_short.append(
|
||
(df["downtrend"] == 1)
|
||
& (df["bearish_engulf"] == 1)
|
||
)
|
||
|
||
# C) 下降趋势 + Inside Bar 向下突破
|
||
conditions_short.append(
|
||
(df["downtrend"] == 1)
|
||
& (df["inside_break_down"] == 1)
|
||
& (df["vol_expand"] == 1)
|
||
)
|
||
|
||
# D) 跌破 Swing Low + 放量
|
||
conditions_short.append(
|
||
(df["break_swing_low"] == 1)
|
||
& (df["vol_expand"] == 1)
|
||
)
|
||
|
||
if conditions_short:
|
||
import pandas as pd
|
||
combined = pd.concat(conditions_short, axis=1).any(axis=1)
|
||
dataframe.loc[combined, "enter_short"] = 1
|
||
|
||
return dataframe
|
||
|
||
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
"""
|
||
出场条件 — 纯价格行为
|
||
|
||
做多出场：
|
||
- 出现看跌吞没
|
||
- 出现看跌 Pin Bar
|
||
- 市场结构转为下降趋势
|
||
- 跌破前一个 Swing Low
|
||
|
||
做空出场（镜像）
|
||
"""
|
||
df = dataframe
|
||
|
||
# 做多出场
|
||
dataframe.loc[
|
||
(df["bearish_engulf"] == 1)
|
||
| (df["bearish_pin"] == 1)
|
||
| (df["downtrend"] == 1)
|
||
| (df["break_swing_low"] == 1),
|
||
"exit_long",
|
||
] = 1
|
||
|
||
# 做空出场
|
||
dataframe.loc[
|
||
(df["bullish_engulf"] == 1)
|
||
| (df["bullish_pin"] == 1)
|
||
| (df["uptrend"] == 1)
|
||
| (df["break_swing_high"] == 1),
|
||
"exit_short",
|
||
] = 1
|
||
|
||
return dataframe
|
||
|
||
def custom_exit(
|
||
self,
|
||
pair: str,
|
||
trade,
|
||
current_time,
|
||
current_rate: float,
|
||
current_profit: float,
|
||
**kwargs,
|
||
):
|
||
"""
|
||
自定义出场 — 基于价格行为的动态止盈
|
||
|
||
1. 利润 > 2% 且出现反转K线 → 锁利
|
||
2. 持仓超过 2 小时且利润 < 0.3% → 超时退出（行情没走出来）
|
||
"""
|
||
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
||
if dataframe.empty or len(dataframe) < 2:
|
||
return None
|
||
|
||
last = dataframe.iloc[-1]
|
||
trade_duration = (current_time - trade.open_date_utc).total_seconds() / 60
|
||
|
||
# 1. 有利润 + 反转K线 → 锁利
|
||
if not trade.is_short:
|
||
if current_profit > 0.02:
|
||
if last.get("bearish_pin", 0) == 1 or last.get("bearish_engulf", 0) == 1:
|
||
return "reversal_signal_tp"
|
||
if current_profit > 0.035:
|
||
# 大利润时，任何阴线都考虑锁利
|
||
if last.get("is_bear", 0) == 1 and last.get("body_ratio", 0) > 0.6:
|
||
return "strong_bear_candle_tp"
|
||
else:
|
||
if current_profit > 0.02:
|
||
if last.get("bullish_pin", 0) == 1 or last.get("bullish_engulf", 0) == 1:
|
||
return "reversal_signal_tp"
|
||
if current_profit > 0.035:
|
||
if last.get("is_bull", 0) == 1 and last.get("body_ratio", 0) > 0.6:
|
||
return "strong_bull_candle_tp"
|
||
|
||
# 2. 超时退出 — 行情没走出来
|
||
if trade_duration > 120 and current_profit < 0.003:
|
||
return "timeout_exit"
|
||
|
||
return None
|
||
|
||
def leverage(
|
||
self,
|
||
pair: str,
|
||
current_time,
|
||
current_rate: float,
|
||
proposed_leverage: float,
|
||
max_leverage: float,
|
||
entry_tag: str | None,
|
||
side: str,
|
||
**kwargs,
|
||
) -> float:
|
||
return 3.0
|